SpringBoot整合RabbitMQ实战教程附死信交换机

<!-- RabbitMQ -->

< dependency >

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

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

     < version >2.7.0</ version >

</ dependency >

spring:

   rabbitmq:

     host: 127.0.0.1

     port: 5672

     username: admin

     password: admin

     virtual-host: admin_host

     publisher-confirm-type: correlated

     publisher-returns: true

     listener:

       simple:

         acknowledge-mode: manual

         retry:

           enabled: true    #开启失败重试

           max-attempts: 3    #最大重试次数

           initial-interval: 1000  #重试间隔时间 毫秒

package com.example.rabitmqdemo.mydemo.config;

import lombok.extern.slf4j.Slf4j;

import org.springframework.amqp.core.*;

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

import org.springframework.context.annotation.Bean;

import org.springframework.stereotype.Component;

import java.util.HashMap;

import java.util.Map;

/**

  * Broker:它提供一种传输服务,它的角色就是维护一条从生产者到消费者的路线，保证数据能按照指定的方式进行传输,

  * Exchange：消息交换机,它指定消息按什么规则,路由到哪个队列。

  * Queue:消息的载体,每个消息都会被投到一个或多个队列。

  * Binding:绑定，它的作用就是把exchange和queue按照路由规则绑定起来.

  * Routing Key:路由关键字,exchange根据这个关键字进行消息投递。

  * vhost:虚拟主机,一个broker里可以有多个vhost，用作不同用户的权限分离。

  * Producer:消息生产者,就是投递消息的程序.

  * Consumer:消息消费者,就是接受消息的程序.

  * Channel:消息通道,在客户端的每个连接里,可建立多个channel.

  */

@Slf4j

@Component

public class RabbitConfig {

     //业务交换机

     public static final String EXCHANGE_PHCP = "phcp" ;

     //业务队列1

     public static final String QUEUE_COMPANY = "company" ;

     //业务队列1的key

     public static final String ROUTINGKEY_COMPANY = "companyKey" ;

     //业务队列2

     public static final String QUEUE_PROJECT = "project" ;

     //业务队列2的key

     public static final String ROUTINGKEY_PROJECT = "projectKey" ;

     //死信交换机

     public static final String EXCHANGE_PHCP_DEAD = "phcp_dead" ;

     //死信队列1

     public static final String QUEUE_COMPANY_DEAD = "company_dead" ;

     //死信队列2

     public static final String QUEUE_PROJECT_DEAD = "project_dead" ;

     //死信队列1的key

     public static final String ROUTINGKEY_COMPANY_DEAD = "companyKey_dead" ;

     //死信队列2的key

     public static final String ROUTINGKEY_PROJECT_DEAD = "projectKey_dead" ;

//    /**

//     * 解决重复确认报错问题，如果没有报错的话，就不用启用这个

//     *

//     * @param connectionFactory

//     * @return

//     */

//    @Bean

//    public RabbitListenerContainerFactory<?> rabbitListenerContainerFactory(ConnectionFactory connectionFactory) {

//        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();

//        factory.setConnectionFactory(connectionFactory);

//        factory.setMessageConverter(new Jackson2JsonMessageConverter());

//        factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);

//        return factory;

//    }

     /**

      * 声明业务交换机

      * 1. 设置交换机类型

      * 2. 将队列绑定到交换机

      * FanoutExchange: 将消息分发到所有的绑定队列，无routingkey的概念

      * HeadersExchange ：通过添加属性key-value匹配

      * DirectExchange:按照routingkey分发到指定队列

      * TopicExchange:多关键字匹配

      */

     @Bean ( "exchangePhcp" )

     public DirectExchange exchangePhcp() {

         return new DirectExchange(EXCHANGE_PHCP);

     }

      * 声明死信交换机

     @Bean ( "exchangePhcpDead" )

     public DirectExchange exchangePhcpDead() {

         return new DirectExchange(EXCHANGE_PHCP_DEAD);

      * 声明业务队列 1

      *

      * @return

     @Bean ( "queueCompany" )

     public Queue queueCompany() {

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

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

         //绑定该队列到死信交换机的队列1

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

         return QueueBuilder.durable(QUEUE_COMPANY).withArguments(arguments).build();

      * 声明业务队列 2

     @Bean ( "queueProject" )

     public Queue queueProject() {

         //绑定该队列到死信交换机的队列2

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

         return QueueBuilder.durable(QUEUE_PROJECT).withArguments(arguments).build();

      * 声明死信队列 1

     @Bean ( "queueCompanyDead" )

     public Queue queueCompanyDead() {

         return new Queue(QUEUE_COMPANY_DEAD);

      * 声明死信队列 2

     @Bean ( "queueProjectDead" )

     public Queue queueProjectDead() {

         return new Queue(QUEUE_PROJECT_DEAD);

      * 绑定业务队列 1 和业务交换机

      * @param queue

      * @param directExchange

     @Bean

     public Binding bindingQueueCompany( @Qualifier ( "queueCompany" ) Queue queue, @Qualifier ( "exchangePhcp" ) DirectExchange directExchange) {

         return BindingBuilder.bind(queue).to(directExchange).with(RabbitConfig.ROUTINGKEY_COMPANY);

      * 绑定业务队列 2 和业务交换机

     public Binding bindingQueueProject( @Qualifier ( "queueProject" ) Queue queue, @Qualifier ( "exchangePhcp" ) DirectExchange directExchange) {

         return BindingBuilder.bind(queue).to(directExchange).with(RabbitConfig.ROUTINGKEY_PROJECT);

      * 绑定死信队列 1 和死信交换机

     public Binding bindingQueueCompanyDead( @Qualifier ( "queueCompanyDead" ) Queue queue, @Qualifier ( "exchangePhcpDead" ) DirectExchange directExchange) {

         return BindingBuilder.bind(queue).to(directExchange).with(RabbitConfig.ROUTINGKEY_COMPANY_DEAD);

      * 绑定死信队列 2 和死信交换机

     public Binding bindingQueueProjectDead( @Qualifier ( "queueProjectDead" ) Queue queue, @Qualifier ( "exchangePhcpDead" ) DirectExchange directExchange) {

         return BindingBuilder.bind(queue).to(directExchange).with(RabbitConfig.ROUTINGKEY_PROJECT_DEAD);

}

package com.example.rabitmqdemo.mydemo.producer;

import com.example.rabitmqdemo.mydemo.config.RabbitConfig;

import lombok.extern.slf4j.Slf4j;

import org.springframework.amqp.core.*;

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

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

import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;

import javax.annotation.Resource;

import java.nio.charset.StandardCharsets;

import java.util.UUID;

@Component

@Slf4j

public class RabbltProducer implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnsCallback{

     @Resource

     private RabbitTemplate rabbitTemplate;

     /**

      * 初始化消息确认函数

      */

     @PostConstruct

     public void init() {

         rabbitTemplate.setConfirmCallback( this );

         rabbitTemplate.setReturnsCallback( this );

         rabbitTemplate.setMandatory( true );

     }

     /**

      * 发送消息服务器确认函数

      * @param correlationData

      * @param ack

      * @param cause

      */

     @Override

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

         if (ack) {

             System.out.println( "消息发送成功" + correlationData);

         } else {

             System.out.println( "消息发送失败:" + cause);

         }

     }

     /**

      * 消息发送失败，消息回调函数

      * @param returnedMessage

      */

     @Override

     public void returnedMessage(ReturnedMessage returnedMessage) {

         String str = new String(returnedMessage.getMessage().getBody());

         System.out.println( "消息发送失败：" + str);

     }

     /**

      * 处理消息发送到队列1

      * @param str

      */

     public void sendCompany(String str){

         MessageProperties messageProperties = new MessageProperties();

         messageProperties.setContentType( "application/json" );

         Message message = new Message(str.getBytes(StandardCharsets.UTF_8),messageProperties);

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

         this .rabbitTemplate.convertAndSend(RabbitConfig.EXCHANGE_PHCP,RabbitConfig.ROUTINGKEY_COMPANY,message,correlationData);

         //也可以用下面的方式

         //CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());

         //this.rabbitTemplate.convertAndSend(RabbitConfig.EXCHANGE_PHCP,RabbitConfig.ROUTINGKEY_COMPANY,str,correlationData);

     }

     /**

      * 处理消息发送到队列2

      * @param str

      */

     public void sendProject(String str){

         MessageProperties messageProperties = new MessageProperties();

         messageProperties.setContentType( "application/json" );

         Message message = new Message(str.getBytes(StandardCharsets.UTF_8),messageProperties);

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

         this .rabbitTemplate.convertAndSend(RabbitConfig.EXCHANGE_PHCP,RabbitConfig.ROUTINGKEY_PROJECT,message,correlationData);

         //也可以用下面的方式

         //CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());

         //this.rabbitTemplate.convertAndSend(RabbitConfig.EXCHANGE_PHCP,RabbitConfig.ROUTINGKEY_PROJECT,str,correlationData);

     }

}

package com.example.rabitmqdemo.mydemo.consumer;

import com.rabbitmq.client.Channel;

import lombok.extern.slf4j.Slf4j;

import org.springframework.amqp.core.Message;

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

import org.springframework.stereotype.Component;

import java.io.IOException;

/**

  * 监听业务交换机

  * @author JeWang

  */

@Component

@Slf4j

public class RabbitConsumer {

     /**

      * 监听业务队列1

      * @param message

      * @param channel

      * @throws IOException

      */

     @RabbitListener (queues = "company" )

     public void company(Message message, Channel channel) throws IOException {

         try {

             System.out.println( "次数" + message.getMessageProperties().getDeliveryTag());

             channel.basicQos( 1 );

             Thread.sleep( 2000 );

             String s = new String(message.getBody());

             log.info( "处理消息" +s);

             //下面两行是尝试手动抛出异常，用来测试重试次数和发送到死信交换机

             //String str = null;

             //str.split("1");

             //处理成功，确认应答

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

         } catch (Exception e){

             log.error( "处理消息时发生异常：" +e.getMessage());

             Boolean redelivered = message.getMessageProperties().getRedelivered();

             if (redelivered){

                 log.error( "异常重试次数已到达设置次数，将发送到死信交换机" );

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

             } else {

                 log.error( "消息即将返回队列处理重试" );

                 channel.basicNack(message.getMessageProperties().getDeliveryTag(), false , true );

             }

         }

     }

     /**

      * 监听业务队列2

      * @param message

      * @param channel

      * @throws IOException

      */

     @RabbitListener (queues = "project" )

     public void project(Message message, Channel channel) throws IOException {

         try {

             System.out.println( "次数" + message.getMessageProperties().getDeliveryTag());

             channel.basicQos( 1 );

             Thread.sleep( 2000 );

             String s = new String(message.getBody());

             log.info( "处理消息" +s);

             //下面两行是尝试手动抛出异常，用来测试重试次数和发送到死信交换机

             //String str = null;

             //str.split("1");

             //处理成功，确认应答

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

         } catch (Exception e){

             log.error( "处理消息时发生异常：" +e.getMessage());

             Boolean redelivered = message.getMessageProperties().getRedelivered();

             if (redelivered){

                 log.error( "异常重试次数已到达设置次数，将发送到死信交换机" );

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

             } else {

                 log.error( "消息即将返回队列处理重试" );

                 channel.basicNack(message.getMessageProperties().getDeliveryTag(), false , true );

             }

         }

     }

}

package com.example.rabitmqdemo.mydemo.consumer;

import com.rabbitmq.client.Channel;

import lombok.extern.slf4j.Slf4j;

import org.springframework.amqp.core.Message;

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

import org.springframework.stereotype.Component;

import java.io.IOException;

/**

  * 监听死信交换机

  * @author JeWang

  */

@Component

@Slf4j

public class RabbitConsumerDead {

     /**

      * 处理死信队列1

      * @param message

      * @param channel

      * @throws IOException

      */

     @RabbitListener (queues = "company_dead" )

     public void company_dead(Message message, Channel channel) throws IOException {

         try {

             channel.basicQos( 1 );

             String s = new String(message.getBody());

             log.info( "处理死信" +s);

             //在此处记录到数据库、报警之类的操作

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

         } catch (Exception e){

             log.error( "接收异常：" +e.getMessage());

         }

     }

     /**

      * 处理死信队列2

      * @param message

      * @param channel

      * @throws IOException

      */

     @RabbitListener (queues = "project_dead" )

     public void project_dead(Message message, Channel channel) throws IOException {

         try {

             channel.basicQos( 1 );

             String s = new String(message.getBody());

             log.info( "处理死信" +s);

             //在此处记录到数据库、报警之类的操作

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

         } catch (Exception e){

             log.error( "接收异常：" +e.getMessage());

         }

     }

}

package com.example.rabitmqdemo.mydemo.controller;

import com.example.rabitmqdemo.mydemo.producer.RabbltProducer;

import lombok.extern.slf4j.Slf4j;

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

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

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

import javax.annotation.Resource;

@RequestMapping ( "/def" )

@RestController

@Slf4j

public class MsgController {

     @Resource

     private RabbltProducer rabbltProducer;

     @RequestMapping ( "/handleCompany" )

     public void handleCompany( @RequestBody String jsonStr){

         rabbltProducer.sendCompany(jsonStr);

     }

}