SpringBoot集成RabbitMQ(注解+手動確認)

 

1.pom文件

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

2.yml配置文件

spring:
 #MQ配置
  rabbitmq:
    addresses: 127.0.0.1
    port: 5672
    username: adminmq
    password: 123456
    publisher-confirms: true
    publisher-returns: true
    template:
      retry:
        enabled: true
      mandatory: true
    listener:
      simple:
        acknowledge-mode: manual
        #並發消費者初始化值
        concurrency: 10
        #並發消費者的最大值
        max-concurrency: 20
        #每個消費者每次監聽時可拉取處理的消息數量
        prefetch: 5
     
      direct:
        retry:
          enabled: true
          max-attempts: 1

3.消費者代碼(手動確認)

/**
 * msgByte:報文頭加報文體
 * channel和message  消息確認機制
 * queuesToDeclare = @Queue("${queueropertie.queue-name}")
 */
//點對點
//@RabbitListener(queuesToDeclare=@Queue(QueueAndExchangeProperties.afsendfirQueue))
//發布訂閱
@RabbitListener(bindings = @QueueBinding(

            exchange = @Exchange(value = "${queueropertie.exchange}",durable = "true",type = "direct"),
            value = @Queue(value = "${queueropertie.queue-name}",durable = "true"),
            key = "${queueropertie.exchangekey}"
    ))
    @RabbitHandler
    public void monitoringMethod(byte[] msgByte, Channel channel, Message message) throws IOException {
        Map<String, Object> logMap = new ConcurrentHashMap<>();
        try {
            //消息確認
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
            monotoringInsertDB(msgByte,new ConcurrentHashMap<>());
        } catch (Exception e) {
            //失敗后消息被確認
            channel.basicReject(message.getMessageProperties().getDeliveryTag(), false);
            commonRabbitService.insertLogError(logMap, ERROR_104, e.getMessage());
            LOGGER.error("mq接收消息失敗",e);
        }
    }

4.生產者

@Autowired
RabbitTemplate rabbitTemplate;
//發布訂閱
rabbitTemplate.convertAndSend(QueueAndExchangeProperties.afMQExchange, QueueAndExchangeProperties.afIcf, msgStr.getBytes());
//點對點
//rabbitTemplate.convertAndSend(QueueAndExchangeProperties.afsendsecQueue,msgStr);