springboot2.5.6集成RabbitMq实现Topic主题模式(推荐)

server:

   port: 8184

spring:

   application:

     name: rabbitmq-demo

   rabbitmq:

     host: 127.0 . 0.1 # ip地址

     port: 5672

     username: admin # 连接账号

     password: 123456 # 连接密码

     template:

       retry:

         enabled: true # 开启失败重试

         initial-interval: 10000ms # 第一次重试的间隔时长

         max-interval: 300000ms # 最长重试间隔，超过这个间隔将不再重试

         multiplier: 2 # 下次重试间隔的倍数，此处是 2 即下次重试间隔是上次的 2 倍

       exchange: topic.exchange # 缺省的交换机名称，此处配置后，发送消息如果不指定交换机就会使用这个

     publisher-confirm-type: correlated # 生产者确认机制，确保消息会正确发送，如果发送失败会有错误回执，从而触发重试

     publisher-returns: true

     listener:

       type: simple

       simple:

         acknowledge-mode: manual

         prefetch: 1 # 限制每次发送一条数据。

         concurrency: 3 # 同一个队列启动几个消费者

         max-concurrency: 3 # 启动消费者最大数量

         # 重试策略相关配置

         retry:

           enabled: true # 是否支持重试

           max-attempts: 5

           stateless: false

           multiplier: 1.0 # 时间策略乘数因子

           initial-interval: 1000ms

           max-interval: 10000ms

         default -requeue-rejected: true

<!-- rabbitmq -->

         <dependency>

             <groupId>org.springframework.boot</groupId>

             <artifactId>spring-boot-starter-amqp</artifactId>

         </dependency>

/**

  * @author kkp

  * @ClassName RabbitMqConstants

  * @date 2021/11/3 14:16

  * @Description

  */

public class RabbitMqConstants {

     public final static String TEST1_QUEUE = "test1-queue" ;

     public final static String TEST2_QUEUE = "test2-queue" ;

     public final static String EXCHANGE_NAME = "test.topic.exchange" ;

     /**

      * routingKey1

      */

     public final static String TOPIC_TEST1_ROUTINGKEY = "topic.test1.*" ;

     public final static String TOPIC_TEST1_ROUTINGKEY_TEST = "topic.test1.test" ;

     /**

      * routingKey1

      */

     public final static String TOPIC_TEST2_ROUTINGKEY = "topic.test2.*" ;

     public final static String TOPIC_TEST2_ROUTINGKEY_TEST = "topic.test2.test" ;

}

import com.example.demo.common.RabbitMqConstants;

import lombok.extern.slf4j.Slf4j;

import org.springframework.amqp.core.*;

import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;

import org.springframework.amqp.rabbit.core.RabbitTemplate;

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.beans.factory.annotation.Qualifier;

import org.springframework.beans.factory.config.ConfigurableBeanFactory;

import org.springframework.context.annotation.Bean;

import org.springframework.context.annotation.Configuration;

import org.springframework.context.annotation.Scope;

/**

  * @author kkp

  * @ClassName RabbitMqConfig

  * @date 2021/11/3 14:16

  * @Description

  */

@Slf4j

@Configuration

public class RabbitMqConfig {

     @Autowired

     private CachingConnectionFactory connectionFactory;

     /**

      *  声明交换机

      */

     @Bean (RabbitMqConstants.EXCHANGE_NAME)

     public Exchange exchange(){

  //durable(true) 持久化，mq重启之后交换机还在

         // Topic模式

         //return ExchangeBuilder.topicExchange(RabbitMqConstants.EXCHANGE_NAME).durable(true).build();

         //发布订阅模式

         return ExchangeBuilder.fanoutExchange(RabbitMqConstants.EXCHANGE_NAME).durable( true ).build();

     }

     /**

      *  声明队列

      *  new Queue(QUEUE_EMAIL,true,false,false)

      *  durable="true" 持久化 rabbitmq重启的时候不需要创建新的队列

      *  auto-delete 表示消息队列没有在使用时将被自动删除 默认是false

      *  exclusive  表示该消息队列是否只在当前connection生效,默认是false

      */

     @Bean (RabbitMqConstants.TEST1_QUEUE)

     public Queue esQueue() {

         return new Queue(RabbitMqConstants.TEST1_QUEUE);

     }

     /**

      *  声明队列

      */

     @Bean (RabbitMqConstants.TEST2_QUEUE)

     public Queue gitalkQueue() {

         return new Queue(RabbitMqConstants.TEST2_QUEUE);

     }

     /**

      *  TEST1_QUEUE队列绑定交换机，指定routingKey

      */

     @Bean

     public Binding bindingEs( @Qualifier (RabbitMqConstants.TEST1_QUEUE) Queue queue,

                              @Qualifier (RabbitMqConstants.EXCHANGE_NAME) Exchange exchange) {

         return BindingBuilder.bind(queue).to(exchange).with(RabbitMqConstants.TOPIC_TEST1_ROUTINGKEY).noargs();

     }

     /**

      *  TEST2_QUEUE队列绑定交换机，指定routingKey

      */

     @Bean

     public Binding bindingGitalk( @Qualifier (RabbitMqConstants.TEST2_QUEUE) Queue queue,

                                  @Qualifier (RabbitMqConstants.EXCHANGE_NAME) Exchange exchange) {

         return BindingBuilder.bind(queue).to(exchange).with(RabbitMqConstants.TOPIC_TEST2_ROUTINGKEY).noargs();

     }

     /**

      * 如果需要在生产者需要消息发送后的回调，

      * 需要对rabbitTemplate设置ConfirmCallback对象，

      * 由于不同的生产者需要对应不同的ConfirmCallback，

      * 如果rabbitTemplate设置为单例bean，

      * 则所有的rabbitTemplate实际的ConfirmCallback为最后一次申明的ConfirmCallback。

      * @return

      */

     @Bean

     @Scope (ConfigurableBeanFactory.SCOPE_PROTOTYPE)

     public RabbitTemplate rabbitTemplate() {

          RabbitTemplate template = new RabbitTemplate(connectionFactory);

         return template;

     }

}

import lombok.extern.slf4j.Slf4j;

import org.springframework.amqp.core.Message;

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;

import java.util.UUID;

/**

  * @author kkp

  * @ClassName RabbitMqUtils

  * @date 2021/11/3 14:21

  * @Description

  */

@Slf4j

@Component

public class RabbitMqUtils implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnCallback{

     private RabbitTemplate rabbitTemplate;

     /**

      * 构造方法注入

      */

     @Autowired

     public RabbitMqUtils(RabbitTemplate rabbitTemplate) {

         this .rabbitTemplate = rabbitTemplate;

         //这是是设置回调能收到发送到响应

         rabbitTemplate.setConfirmCallback( this );

         //如果设置备份队列则不起作用

         rabbitTemplate.setMandatory( true );

         rabbitTemplate.setReturnCallback( this );

     }

     /**

      * 回调确认

      */

     @Override

     public void confirm(CorrelationData correlationData, boolean ack, String cause) {

         if (ack){

             log.info( "消息发送成功：correlationData({}),ack({}),cause({})" ,correlationData,ack,cause);

         } else {

             log.info( "消息发送失败：correlationData({}),ack({}),cause({})" ,correlationData,ack,cause);

         }

     }

     /**

      * 消息发送到转换器的时候没有对列,配置了备份对列该回调则不生效

      * @param message

      * @param replyCode

      * @param replyText

      * @param exchange

      * @param routingKey

      */

     @Override

     public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {

         log.info( "消息丢失：exchange({}),route({}),replyCode({}),replyText({}),message:{}" ,exchange,routingKey,replyCode,replyText,message);

     }

     /**

      * 发送到指定Queue

      * @param queueName

      * @param obj

      */

     public void send(String queueName, Object obj){

         CorrelationData correlationId = new CorrelationData(UUID.randomUUID().toString());

         this .rabbitTemplate.convertAndSend(queueName, obj, correlationId);

     }

     /**

      * 1、交换机名称

      * 2、routingKey

      * 3、消息内容

      */

     public void sendByRoutingKey(String exChange, String routingKey, Object obj){

         CorrelationData correlationId = new CorrelationData(UUID.randomUUID().toString());

         this .rabbitTemplate.convertAndSend(exChange, routingKey, obj, correlationId);

     }

}

public interface TestService {

     String sendTest1(String content);

     String sendTest2(String content);

}

import com.example.demo.common.RabbitMqConstants;

import com.example.demo.util.RabbitMqUtils;

import lombok.extern.slf4j.Slf4j;

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.stereotype.Service;

/**

  * @author kkp

  * @ClassName TestServiceImpl

  * @date 2021/11/3 14:24

  * @Description

  */

@Service

@Slf4j

public class TestServiceImpl implements TestService {

     @Autowired

     private RabbitMqUtils rabbitMqUtils;

     @Override

     public String sendTest1(String content) {

         rabbitMqUtils.sendByRoutingKey(RabbitMqConstants.EXCHANGE_NAME,

                 RabbitMqConstants.TOPIC_TEST1_ROUTINGKEY_TEST, content);

         log.info(RabbitMqConstants.TOPIC_TEST1_ROUTINGKEY_TEST+ "***************发送成功*****************" );

         return "发送成功！" ;

     }

     @Override

     public String sendTest2(String content) {

         rabbitMqUtils.sendByRoutingKey(RabbitMqConstants.EXCHANGE_NAME,

                 RabbitMqConstants.TOPIC_TEST2_ROUTINGKEY_TEST, content);

         log.info(RabbitMqConstants.TOPIC_TEST2_ROUTINGKEY_TEST+ "***************发送成功*****************" );

         return "发送成功！" ;

     }

}

import com.example.demo.common.RabbitMqConstants;

import lombok.extern.slf4j.Slf4j;

import org.springframework.amqp.core.ExchangeTypes;

import org.springframework.amqp.core.Message;

import org.springframework.amqp.rabbit.annotation.Exchange;

import org.springframework.amqp.rabbit.annotation.Queue;

import org.springframework.amqp.rabbit.annotation.QueueBinding;

import org.springframework.amqp.rabbit.annotation.RabbitListener;

import org.springframework.stereotype.Component;

import com.rabbitmq.client.Channel;

/**

  * @author kkp

  * @ClassName RabbitMqListener

  * @date 2021/11/3 14:22

  * @Description

  */

@Slf4j

@Component

public class RabbitMqListener {

     @RabbitListener (queues = RabbitMqConstants.TEST1_QUEUE)

     public void test1Consumer(Message message, Channel channel) {

         try {

             //手动确认消息已经被消费

             channel.basicAck(message.getMessageProperties().getDeliveryTag(), false );

             log.info( "Counsoum1消费消息：" + message.toString() + "。成功！" );

         } catch (Exception e) {

             e.printStackTrace();

             log.info( "Counsoum1消费消息：" + message.toString() + "。失败！" );

         }

     }

     @RabbitListener (queues = RabbitMqConstants.TEST2_QUEUE)

     public void test2Consumer(Message message, Channel channel) {

         try {

             //手动确认消息已经被消费

             channel.basicAck(message.getMessageProperties().getDeliveryTag(), false );

             log.info( "Counsoum2消费消息：" + message.toString() + "。成功！" );

         } catch (Exception e) {

             e.printStackTrace();

             log.info( "Counsoum2消费消息：" + message.toString() + "。失败！" );

         }

     }

}

import com.example.demo.server.TestService;

import jdk.nashorn.internal.objects.annotations.Getter;

import lombok.extern.slf4j.Slf4j;

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.web.bind.annotation.*;

import java.util.Map;

/**

  * @author kkp

  * @ClassName TestController

  * @date 2021/11/3 14:25

  * @Description

  */

@Slf4j

@RestController

@RequestMapping ( "/enterprise" )

public class TestController {

     @Autowired

     private TestService testService;

     @GetMapping ( "/finance" )

     public String hello3( @RequestParam (required = false ) Map<String, Object> params) {

         return testService.sendTest2(params.get( "entId" ).toString());

     }

     /**

      * 发送消息test2

      * @param content

      * @return

      */

     @PostMapping (value = "/finance2" )

     public String sendTest2( @RequestBody String content) {

         return testService.sendTest2(content);

     }

}