文章详情

短信预约-IT技能 免费直播动态提醒

请输入下面的图形验证码

提交验证

短信预约提醒成功

RabbitMQ 3.9.7 镜像模式集群与Springboot 2.5.5 整合

2024-04-02 19:55

关注

1. 概述

老话说的好:做人要懂得变通,善于思考,有时稍微转个弯,也许问题就解决了。

言归正传,之前我们聊了 RabbitMQ 3.9.7 镜像模式集群的搭建,今天我们来聊聊 RabbitMQ 3.9.7 镜像模式集群与Springboot 2.5.5 整合。

2. 场景说明

服务器A IP:192.168.1.22

服务器B IP:192.168.1.8

服务器C IP:192.168.1.144

此三台服务器上已搭建好了 RabbitMQ镜像模式集群,镜像模式集群的搭建,可参见我的上一篇文章。

3. 与Springboot的整合

3.1 引入依赖


<parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.5.5</version>
        <relativePath/> <!-- lookup parent from repository -->
    </parent>

 <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-amqp</artifactId>
        </dependency>

3.2 生产服务配置


spring:
  rabbitmq:
    addresses: 192.168.1.22:5672,192.168.1.8:5672,192.168.1.144:5672
    username: guest
    password: guest
    virtual-host: /
    connection-timeout: 16000

    # 启用消息确认模式
    publisher-confirm-type: correlated

    # 启用 return 消息模式
    publisher-returns: true
    template:
      mandatory: true

3.3 生产服务代码


import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;

import java.util.Map;

@Component
public class Producer {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    
    final RabbitTemplate.ConfirmCallback confirmCallback = new RabbitTemplate.ConfirmCallback() {

        @Override
        public void confirm(CorrelationData correlationData, boolean ack, String cause) {
            // correlationData 唯一标识
            // ack mq是否收到消息
            // cause 失败原因
            System.out.println("correlationData:" + correlationData.getId());
            System.out.println("ack:" + ack);
            System.out.println("cause:" + cause);

        }
    };

    
    public void sendMessage(String messageBody, Map<String, Object> headers, String id) throws Exception {

        MessageHeaders messageHeaders = new MessageHeaders(headers);

        Message<String> message = MessageBuilder.createMessage(messageBody, messageHeaders);

        rabbitTemplate.setConfirmCallback(confirmCallback);

        String exchangeName = "exchange-hello";
        String routingKey = "test.123";

        CorrelationData correlationData = new CorrelationData(id);

        rabbitTemplate.convertAndSend(exchangeName, routingKey, message, new MessagePostProcessor() {

            
            @Override
            public org.springframework.amqp.core.Message postProcessMessage(org.springframework.amqp.core.Message message) throws AmqpException {
                return message;
            }
        }, correlationData);
    }
}

3.4 消费服务配置


spring:
  rabbitmq:
    addresses: 192.168.1.22:5672,192.168.1.8:5672,192.168.1.144:5672
    username: guest
    password: guest
    virtual-host: /
    connection-timeout: 16000

    listener:
      simple:
        # 设置为手工ACK
        acknowledge-mode: manual
        concurrency: 5
        prefetch: 1
        max-concurrency: 10

3.5 消费服务代码


import com.rabbitmq.client.Channel;
import org.springframework.amqp.rabbit.annotation.*;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.messaging.Message;
import org.springframework.stereotype.Component;

@Component
public class Consumer {

    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(value = "queue-hello", durable = "true"),
            exchange = @Exchange(value = "exchange-hello" , durable = "true", type = "topic"),
            key = "test.*"
    ))
    @RabbitHandler
    public void onMessage(Message message, Channel channel) throws Exception {

        System.out.println("收到消息:" + message.getPayload());

        Long deliveryTag = (Long)message.getHeaders().get(AmqpHeaders.DELIVERY_TAG);
        channel.basicAck(deliveryTag, false);
    }
}

3.6 Rest 测试代码


@RestController
@RequestMapping("/mq")
public class RabbitmqController {

    @Autowired
    private Producer producer;

    @GetMapping("/sendMessage")
    public String sendMessage(@RequestParam String messageBody, @RequestParam String id) throws Exception {
        Map<String, Object> headers = new HashMap<>();
        producer.sendMessage(messageBody, headers, id);
        return "success";
    }
}

4. 综述

到此这篇关于RabbitMQ 3.9.7 镜像模式集群与Springboot 2.5.5 整合的文章就介绍到这了,更多相关RabbitMQ镜像模式集群内容请搜索编程网以前的文章或继续浏览下面的相关文章希望大家以后多多支持编程网!

阅读原文内容投诉

免责声明:

① 本站未注明“稿件来源”的信息均来自网络整理。其文字、图片和音视频稿件的所属权归原作者所有。本站收集整理出于非商业性的教育和科研之目的,并不意味着本站赞同其观点或证实其内容的真实性。仅作为临时的测试数据,供内部测试之用。本站并未授权任何人以任何方式主动获取本站任何信息。

② 本站未注明“稿件来源”的临时测试数据将在测试完成后最终做删除处理。有问题或投稿请发送至: 邮箱/279061341@qq.com QQ/279061341

软考中级精品资料免费领

  • 历年真题答案解析
  • 备考技巧名师总结
  • 高频考点精准押题
  • 2024年上半年信息系统项目管理师第二批次真题及答案解析(完整版)

    难度     813人已做
    查看
  • 【考后总结】2024年5月26日信息系统项目管理师第2批次考情分析

    难度     354人已做
    查看
  • 【考后总结】2024年5月25日信息系统项目管理师第1批次考情分析

    难度     318人已做
    查看
  • 2024年上半年软考高项第一、二批次真题考点汇总(完整版)

    难度     435人已做
    查看
  • 2024年上半年系统架构设计师考试综合知识真题

    难度     224人已做
    查看

相关文章

发现更多好内容

猜你喜欢

AI推送时光机
位置:首页-资讯-后端开发
咦!没有更多了?去看看其它编程学习网 内容吧
首页课程
资料下载
问答资讯