SpringBoot整合RabbitMQ及生产全场景高级特性实战

< parent >

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

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

     < version >2.1.4.RELEASE</ version >

</ parent >

< dependencies >

     < dependency >

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

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

     </ dependency >

     < dependency >

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

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

     </ dependency >

</ dependencies >

server.port=8090

spring.rabbitmq.host=192.168.168.10

spring.rabbitmq.port=5672

spring.rabbitmq.username=zheng123

spring.rabbitmq.password=zheng123

spring.rabbitmq.virtual-host=/zheng

spring.rabbitmq.publisher-confirms=true

spring.rabbitmq.publisher-returns=true

spring.rabbitmq.listener.direct.acknowledge-mode=manual

@Configuration

public class TopicConfig {

    // 声明队列

    @Bean

    public Queue topicQ1() {

       return new Queue( "topic_sb_mq_q1" );

    }

    // 声明队列并绑定该队列到死信交换机(返回值有两种写法，任选一种都可以)

    // 测试死信需要关闭原队列的监听

    @Bean

    public Queue topicQ2() {

         return QueueBuilder.durable( "topic_sb_mq_q2" )

               .withArgument( "x-dead-letter-exchange" , "topicExchange" )

               .withArgument( "x-dead-letter-routing-key" , "changsha.f" )

               .withArgument( "x-message-ttl" , 10000 )

               .build();

         Map<String,Object> arguments = new HashMap<>( 2 );

         arguments.put( "x-dead-letter-exchange" , "topicExchange" );

         arguments.put( "x-dead-letter-routing-key" , "changsha.f" );

         arguments.put( "x-message-ttl" , 10000 );

         return new Queue( "topic_sb_mq_q2" , true , false , false ,arguments);

    }

    //声明exchange

    @Bean

    public TopicExchange setTopicExchange() {

       return new TopicExchange( "topicExchange" );

    }

    //声明binding，需要声明一个routingKey

    @Bean

    public Binding bindTopicHebei1() {

       return BindingBuilder.bind(topicQ1()).to(setTopicExchange()).with( "changsha.*" );

    }

    @Bean

    public Binding bindTopicHebei2() {

       return BindingBuilder.bind(topicQ2()).to(setTopicExchange()).with( "#.beijing" );

    }

}

@Configuration

public class RabbitMqConfig {

     //定义交换机的名字

     public static final String  EXCHANGE_NAME = "boot_topic_exchange" ;

     //定义队列的名字

     public static final String QUEUE_NAME = "boot_queue" ;

     //1、声明交换机

     @Bean ( "bootExchange" )

     public Exchange bootExchange(){

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

     }

     //2、声明队列

     @Bean ( "bootQueue" )

     public Queue bootQueue(){

         return QueueBuilder.durable(QUEUE_NAME).build();

     }

     //3、队列与交换机进行绑定

     @Bean

     public Binding bindQueueExchange( @Qualifier ( "bootQueue" ) Queue queue, @Qualifier ( "bootExchange" ) Exchange exchange){

         // topic模式兼容广播模式，路由模式。with("#")则类似广播模式匹配所有订阅者；with("boot.1")则类似路由模式匹配指定订阅者

         return BindingBuilder.bind(queue).to(exchange).with( "boot.#" ).noargs();

     }

}

@RestController

public class ProducerController {

     @Autowired

     private RabbitTemplate rabbitTemplate;

     @GetMapping (value= "/topicSend" )

     public Object topicSend(String routingKey,String message) throws AmqpException, UnsupportedEncodingException {

        // 定义 confirm 回调

        rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {

           if (ack) {

              // confirmed

           } else {

              // nack-ed

           }

        });

        rabbitTemplate.setMandatory( true );

        rabbitTemplate.setReturnCallback((msg, replyCode, replyText, exchange, routKey)->{

           // return message

        });

        if ( null == routingKey) {

           routingKey= "changsha.kf" ;

        }

        MessageProperties messageProperties = new MessageProperties();

        messageProperties.setContentType(MessageProperties.CONTENT_TYPE_TEXT_PLAIN);

        //fanout模式只往exchange里发送消息。分发到exchange下的所有queue

        rabbitTemplate.send( "topicExchange" , routingKey, new Message(message.getBytes( "UTF-8" ),messageProperties));

        return "message sended : routingKey >" +routingKey+ ";message > " +message;

     }

}

@Component

public class ConcumerReceiver {

     //topic 模式

     //注意这个模式会有优先匹配原则。例如发送routingKey=hunan.IT,那匹配到hunan.*(hunan.IT,hunan.eco),之后就不会再去匹配*.ITd

     @RabbitListener (queues= "topic_sb_mq_q1" )

     public void topicReceiveq1(String msg,Message message, Channel channel) throws IOException {

        // 消息id

        long deliveryTag = message.getMessageProperties().getDeliveryTag();

        try {

           // message.getBody() todosomething

           // 签收消息

           channel.basicAck(deliveryTag, true );

        } catch (Exception e) {

           // 拒绝签收

           // 第三个参数：requeue：重回队列。如果设置为true，则消息重新回到queue，broker会重新发送该消息给消费端

           channel.basicNack(deliveryTag, true , true );

        }

     }

     @RabbitListener (queues= "topic_sb_mq_q2" )

     public void topicReceiveq2(String message) {

        System.out.println( "Topic模式 topic_sb_mq_q2 received  message : " +message);

     }

}