springboot整合RabbitMQ_springboot rabbitmq config

springboot整合框架思想

利用配置文件配置的参数去构造框架对象注入容器，后续将对象注入到属性中去使用。
在导入依赖，配置完yml文件之后，系统启动就会自动注入一个rabbitmqTemplate。

一、安装RabbitMQ

RabbitMQ官网

1.1、需要先安装Erlang

Erlang官网

1.2、安装RabbitMQ

下载安装包

二、添加依赖

<dependency>  
    <groupId>org.springframework.boot</groupId>  
    <artifactId>spring-boot-starter-amqp</artifactId>  
    <version>2.6.4</version>  
</dependency>
<!--rabbitmq消息队列-->
<dependency>
	<groupId>org.jeecgframework.boot</groupId>
	<artifactId>sd-business-starter-rabbitmq</artifactId>
	<version>1.0.0</version>
</dependency>
1
2
3
4
5
6
7
8
9
10
11

三、YAML文件配置RabbitMQ连接

# 消息队列配置
  rabbitmq:
    host: xx.xx.xx.xx                  #rabbitmq的连接地址
    port: 5672                         #rabbitmq的连接amqp端口号
    virtual-host: /                    #rabbitmq的虚拟host
    username: xxxx                     #rabbitmq的用户名
    password: xxxx        #rabbitmq的密码
    listener:
      simple:
        acknowledge-mode: manual       #确认消息-手动确认
        prefetch: 10                   #消费者每次拉取数量
        retry:
          enabled: true                # 开启重试
          max-attempts: 3              # 重试次数,默认为3次
          initial-interval: 5000       #重试间隔时间（单位毫秒）
          max-interval: 1200000        #重试最大时间间隔（单位毫秒）、
          multiplier: 2                #应用于前一重试间隔的乘法器
    publisher-confirm-type: correlated #发布确认属性配置;新版本已弃用publisher-confirms
    template:
      mandatory: true                  #配合发布确认
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

四、配置交换机、队列、绑定关系

• 配置的队列名称和路由键最好一样。
• messageSendDelayPluginBinding方法的参数名需要和上面定义的绑定交换机、队列的方法名一样。

@Configuration
public class RabbitMqConfig {
  	/**
     * 短信群发立即推送队列、延迟队列所绑定的交换机
     **/
    @Bean
    CustomExchange messageSendDelayPluginDirectExchange() {
        HashMap<String, Object> args = new HashMap<>();
        args.put("x-delayed-type", "direct");
        return new CustomExchange("messageSend.updatePushRecordAndTask.direct", "x-delayed-message", true, false, args); // durable=true交换机持久化
    }

    /**
     * 声明短信群发立即推送队列、延迟队列
     **/
    @Bean
    public Queue messageSendDelayPluginQueue() { // durable=true队列持久化
        return new Queue("school.messageSend.updatePushRecordAndTask", true, false, false);
    }

    /**
     * 短信群发立即推送队列、延迟队列绑定交换机，并指定 routingKey,routingKey最好和queue一样
     **/
    @Bean
    public Binding messageSendDelayPluginBinding(Queue messageSendDelayPluginQueue, CustomExchange messageSendDelayPluginDirectExchange) {
        return BindingBuilder.bind(messageSendDelayPluginQueue)
                .to(messageSendDelayPluginDirectExchange)
                .with("school.messageSend.updatePushRecordAndTask")
                .noargs();
    }
 }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

五、创建生产者

5.1、参数解释

rabbitTemplate.convertAndSend(String exchange, String routingKey, Object message, MessagePostProcessor messagePostProcessor, @Nullable CorrelationData correlationData,Envelope envelope)

5.2、例子

import org.jeecg.modules.admin.massSend.bo.UpdatePushRecordAndTaskBO;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.Message;
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.stereotype.Component;

@Component
public class MessageSendUpdatePushRecordAndTaskProducer {
	@Autowired
	RabbitTemplate rabbitTemplate;

	public void sendMessage(UpdatePushRecordAndTaskBO updatePushRecordAndTaskBO) {
		CorrelationData correlationData = new CorrelationData();
		correlationData.setId(updatePushRecordAndTaskBO.getMassSendNo());
		//给延时插件队列发送消息
		rabbitTemplate.convertAndSend("messageSend.updatePushRecordAndTask.direct", "school.messageSend.updatePushRecordAndTask", updatePushRecordAndTaskBO, new MessagePostProcessor() {
					@Override
					public Message postProcessMessage(Message message) throws AmqpException {
						//给消息设置延迟毫秒值，通过给消息设置x-delay头来设置消息从交换机发送到队列的延迟时间，值为0代表不延迟
						message.getMessageProperties().setHeader("x-delay", updatePushRecordAndTaskBO.getDelayTime());
						message.getMessageProperties().setCorrelationId(updatePushRecordAndTaskBO.getSendNo());
						return message;
					}
				}, correlationData);
	}
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

//必须要实现序列号接口
@Data
public class UpdatePushRecordAndTaskBO  implements Serializable {
	@ApiModelProperty(value = "群发编号")
	private java.lang.String sendNo;

	@ApiModelProperty(value = "延迟时间-定时发送使用")
	private Long delayTime;
}
1
2
3
4
5
6
7
8
9

六、创建消费者

6.1、参数解释

消费者handle方法的参数是自定义的，参数存在就会进行注入。

@RabbitListener(queues = “school.imBatchPush.push”, containerFactory = “rabbitListenerContainerFactory”)

handle(ImBatchPushBO imBatchPushBO, Channel channel, @Headers Map<String, Object> map,CorrelationData correlationData)

6.2、消息的手动确认和否认

6.3、例子

import cn.hutool.core.util.ObjectUtil;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.jeecg.common.constant.CommonConstant;
import org.jeecg.common.util.DateUtils;
import org.jeecg.modules.admin.massSend.bo.UpdatePushRecordAndTaskBO;
import org.jeecg.modules.admin.massSend.entity.SdSchoolMessageSendTask;
import org.jeecg.modules.admin.massSend.entity.SdSchoolMessagePushRecord;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.handler.annotation.Headers;
import org.springframework.stereotype.Component;
import java.io.IOException;
import java.util.Map;

@Slf4j
@Component
@RabbitListener(queues = "school.messageSend.updatePushRecordAndTask", containerFactory = "rabbitListenerContainerFactory")
public class MessageSendUpdatePushRecordAndTaskConsumer {

    @RabbitHandler
    public void handle(UpdatePushRecordAndTaskBO updatePushRecordAndTaskBO, Channel channel, @Headers Map<String, Object> map) throws Exception {
        log.info("短信群发更新状态开始: updateTaskBO={}", updatePushRecordAndTaskBO.toString());
        if (map.get("error") != null) {
            System.out.println("错误的消息");
            try {
            	//否认消息，消息会被重新放回消息队列
                channel.basicNack((Long) map.get(AmqpHeaders.DELIVERY_TAG), false, true);      
                return;
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
        try {
            //具体的消费者业务逻辑
            log.info("具体的消费业务逻辑执行了");
            channel.basicAck((Long) map.get(AmqpHeaders.DELIVERY_TAG), false);            //确认消息
        } catch (IOException e) {
            log.error("短信群发更新状态否认消息异常");
        }

    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47

七、使用

直接将生产者像其他类作为一个属性注入到类中，然后像普通的对象调用方法那样调用即可。

    @Autowired
    private MessageMassSendUpdatePushRecordAndTaskProducer messageMassSendUpdatePushRecordAndTaskProducer;
    public void test(){
    	UpdatePushRecordAndTaskBO updatePushRecordAndTaskBO = new UpdatePushRecordAndTaskBO();
    	updatePushRecordAndTaskBO.setMassSendNo(massTask.getSendNo());
    	updatePushRecordAndTaskBO.setDelayTime(0L);
    	messageSendUpdatePushRecordAndTaskProducer.sendMessage(updatePushRecordAndTaskBO);
    }
1
2
3
4
5
6
7
8