SpringCloud 集成 RabbitMQ死信队列_springcloud stream rabbitmq 死信队列

介绍：

   Message：消息，应用程序和服务器之间传送的数据，消息可以非常简单，也可以很复杂。有Properties和Body组成。Properties为外包装，可以对消息进行修饰，比如消息的优先级、延迟等高级特性；Body就是消息体内容。
   Virtual Host：虚拟主机，用于逻辑隔离。一个虚拟主机里面可以有若干个Exchange和Queue，同一个虚拟主机里面不能有相同名称的Exchange或Queue。
   Exchange：交换器，接收消息，按照路由规则将消息路由到一个或者多个队列。如果路由不到，或者返回给生产者，或者直接丢弃。RabbitMQ常用的交换器常用类型有direct、topic、fanout、headers四种，后面详细介绍。
   Binding：绑定，交换器和消息队列之间的虚拟连接，绑定中可以包含一个或者多个RoutingKey。
   RoutingKey：路由键，生产者将消息发送给交换器的时候，会发送一个RoutingKey，用来指定路由规则，这样交换器就知道把消息发送到哪个队列。路由键通常为一个“.”分割的字符串，例如“com.rabbitmq”。
   Queue：消息队列，用来保存消息，供消费者消费。
1
2
3
4
5
6

常用交换器

Direct Exchange （路由模式）

该类型的交换器将所有发送到该交换器的消息被转发到RoutingKey指定的队列中，也就是说路由到BindingKey和RoutingKey完全匹配的队列中。


Topic Exchange （通配符模式）

该类型的交换器将所有发送到Topic Exchange的消息被转发到所有RoutingKey中指定的Topic的队列上面。
Exchange将RoutingKey和某Topic进行模糊匹配，其中“”用来匹配一个词，“#”用于匹配一个或者多个词。例如“com.#”能匹配到“com.rabbitmq.oa”和“com.rabbitmq”；而"login."只能匹配到“com.rabbitmq”。


Fanout Exchange（发布/订阅模式）

该类型不处理路由键，会把所有发送到交换器的消息路由到所有绑定的队列中。优点是转发消息最快，性能最好。

Headers Exchange

该类型的交换器不依赖路由规则来路由消息，而是根据消息内容中的headers属性进行匹配。headers类型交换器性能差，在实际中并不常用。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

本文是通过ROUTING_KEY直接绑定到指定的Queue

jar包依赖

	<!-- rabbitmq依赖 -->
	<dependency>
		<groupId>org.springframework.boot</groupId>
		<artifactId>spring-boot-starter-amqp</artifactId>
	</dependency>
1
2
3
4
5

application.yml 配置

spring: 
   rabbitmq:
      host: 192.168.8.153
      port: 5672
      username: admin
      password: admin
      publisher-confirms: true
      listener:
         simple:
            default-requeue-rejected: false
            retry:
               #重试次数
               max-attempts: 3
               #开启重试机制
               enabled: true
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

RabbitMQ生产者配置信息队列创建

import org.springframework.amqp.core.*;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.HashMap;
import java.util.Map;



/**
 * rabbitmq配置类
 */
@Configuration
public class RabbitConfig {
	


	
    // 创建订单交换机
    public static final String Exchange_Name_ZB_Create = "orderExchange_ZB_Create";
    
	// 创建订单队列
    public static final String Queue_Name_ZB_Create = "orderQueue_ZB_Create";
    
    
	// 创建订单超时队列
    public static final String out_Queue_Name_ZB_Create = "outQueue_ZB_Create";

    
    // 创建订单超时交换机key
    public static final String DEAD_LETTER_REDIRECT_ROUTING_KEY_ZB_Create = "TKEY_R_ZB_Create";
    
    // 创建订单与交换机绑定key
    public static final String DEAD_LETTER_TEST_ROUTING_KEY_ZB_Create = "TDL_KEY_ZB_Create";
    
    
    
    
    /**
     * 创建订单超时队列
     * @return
     */
    @Bean
    public Queue ZBCreateOutQueue() {
        return new Queue(RabbitConfig.out_Queue_Name_ZB_Create, true,false,false);
    }
    

    
    /**
     * 创建订单死信订单交换机
     * @return
     */
    @Bean
    public Exchange ZBCreateDeadLetterExchange() {
        return ExchangeBuilder.directExchange(Exchange_Name_ZB_Create).durable(true).build();
    }
    
    
    /**
     * 创建订单死信订单队列
     * @return
     */
    @Bean
    public Queue ZBCreateDeadLetterQueue() {
    	Map<String, Object> args = new HashMap<>(2);
//      x-dead-letter-exchange    声明  死信队列Exchange
       args.put("x-dead-letter-exchange", Exchange_Name_ZB_Create);
//      x-dead-letter-routing-key    声明 死信队列抛出异常重定向队列的routingKey(TKEY_R)
       args.put("x-dead-letter-routing-key", DEAD_LETTER_REDIRECT_ROUTING_KEY_ZB_Create);
       return QueueBuilder.durable(Queue_Name_ZB_Create).withArguments(args).build();
//        return new Queue(RabbitConfig.Timeout_Trade_Queue_Name, true, false, false); 
    }
    
    
    /**
     * 创建订单死信路由通过 DL_KEY 绑定键绑定到死信队列上.
     *
     * @return the binding
     */
    @Bean
    public Binding ZBCreateDeadLetterBinding() {
        return new Binding(Queue_Name_ZB_Create, Binding.DestinationType.QUEUE, Exchange_Name_ZB_Create, DEAD_LETTER_TEST_ROUTING_KEY_ZB_Create, null);

    }
    
    
    /**
     * 创建订单死信路由通过 KEY_R 绑定键绑定到死信队列上.
     *
     * @return the binding
     */
    @Bean
    public Binding ZBCreateRedirectBinding() {
        return new Binding(out_Queue_Name_ZB_Create, Binding.DestinationType.QUEUE, Exchange_Name_ZB_Create, DEAD_LETTER_REDIRECT_ROUTING_KEY_ZB_Create, null);
    }
    
    
    /**
     * 消息发送格式
     * @return
     */
    @Bean
    MessageConverter jackson2JsonMessageConverter() {
        return new Jackson2JsonMessageConverter();
    }

    
}
    
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
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113

消费者监听死信

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;

@Component
public class RabbitMQConsumer {
	
	private Logger logger = LoggerFactory.getLogger(RabbitMQConsumer.class);
	
	
	@Autowired
	private CrowdsourceService crowdsourceService;
	
	
	/**
	 * 众包创建订单超时监听
	 * @param 
	 */
	@RabbitListener(queuesToDeclare = @Queue("outQueue_ZB_Create"))
	@RabbitHandler
	public void ZBOutMessage(@Payload Crowdsource crowdsource){
			logger.info("-------------------- 噜噜噜噜噜噜--------------------");
			//可根据自己需求调用处理
			crowdsourceService.outZBCreateOrder(crowdsource);
	}
}
	
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

业务代码

	@Autowired
	private AmqpTemplate amqpTemplate;
	
	private MessagePostProcessor messagePostProcessor = message -> {
	       MessageProperties messageProperties = message.getMessageProperties();
	       //设置编码
	       messageProperties.setContentEncoding("utf-8");
	       //设置过期时间10*1000毫秒
	       messageProperties.setExpiration("10000");
	       return message;
	   };

	/**
	* 放入消息队列
	*/
  @Override
  public void push(Integer id) {
  	//此Crowdsource  可根据自己需求传送
    Crowdsource crowdsource = new Crowdsource();
	amqpTemplate.convertSendAndReceive(RabbitConfig.Queue_Name_ZB_Create, crowdsource,messagePostProcessor);
  }


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

RabbitMQ消费者监听

import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;

@Component
public class RabbitMQConsumer {


	@RabbitListener(queues = "orderQueue_ZB_Create")
	@RabbitHandler
	public R ZBCreateOrderMessage(@Payload Crowdsource crowdsource)  throws ParseException{
			logger.info("--------------------噜噜噜噜噜噜--------------------");
			return OrderService.createCrowdOrder(crowdsource);
	}
}


1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21