深入剖析 RabbitMQ —— Spring 框架下实现 AMQP 高级消息队列协议（下）

 1 @Configuration 
 2 public class BindingConfig {
 3 public final static String Queue_First="direct.queue.first";
 4 public final static String Exchange_Name="directExchange";
 5 public final static String Routing_Key_First="directKey1";
 6 
 7 @Bean
 8 public Queue queueFirst(){
 9 return new Queue(this.Queue_First);
10 }
11 
12 @Bean
13 public DirectExchange directExchange(){
14 return new DirectExchange(this.Exchange_Name);
15 }
16 
17 @Bean
18 public Binding bindingExchangeFirst(Queue queueFirst, DirectExchange directExchange){
19 return BindingBuilder.bind(queueFirst).to(directExchange).with(Routing_Key_First);
20 }
21 }
22 
23 @Configuration
24 public class ConnectionConfig {
25 @Value("${spring.rabbitmq.host}")
26 public String host;
27 
28 @Value("${spring.rabbitmq.port}")
29 public int port;
30 
31 @Value("${spring.rabbitmq.username}")
32 public String username;
33 
34 @Value("${spring.rabbitmq.password}")
35 public String password;
36 
37 @Value("${spring.rabbitmq.virtual-host}")
38 public String virtualHost;
39 
40 @Bean
41 public ConnectionFactory getConnectionFactory(){
42 CachingConnectionFactory factory=new CachingConnectionFactory();
43 System.out.println(host);
44 factory.setHost(host);
45 factory.setPort(port);
46 factory.setUsername(username);
47 factory.setPassword(password);
48 factory.setVirtualHost(virtualHost);
49 return factory;
50 }
51 }
52 
53 @Controller
54 @RequestMapping("/producer")
55 public class ProducerController {
56 @Autowired
57 private RabbitTemplate template;
58 
59 @RequestMapping("/send")
60 public void send() {
61 for(int n=0;n<10;n++){ 
62 template.convertAndSend(BindingConfig.Exchange_NAME,BindingConfig.RoutingKey1,"Hello World! "
63 +String.valueOf(n),getCorrelationData());
64 }
65 }
66 
67 private CorrelationData getCorrelationData(){
68 return new CorrelationData(UUID.randomUUID().toString());
69 }
70 }

 1 @Configuration
 2 public class BindingConfig {
 3 //普通队列参数
 4 public final static String Queue_First="direct.queue.first";
 5 public final static String Exchange_Name="directExchange";
 6 public final static String Routing_Key_First="directKey1";
 7 //死信队列参数
 8 public final static String Queue_Dead="direct.queue.dead";
 9 public final static String Exchange_Dead="directDead";
 10 public final static String Routing_Key_Dead="directDeadKey";
 11 
 12 @Bean
 13 public Queue queueFirst(){
 14 Map<String, Object> args=new HashMap<String,Object>();
 15 //声明当前死信的 Exchange
 16 args.put("x-dead-letter-exchange", this.Exchange_Dead);
 17 //声明当前队列的死信路由key
 18 args.put("x-dead-letter-routing-key", this.Routing_Key_Dead);
 19 //把死信队列的参数绑定到当前队列中
 20 return QueueBuilder.durable(Queue_First).withArguments(args).build();
 21 }
 22 
 23 @Bean
 24 public DirectExchange directExchange(){
 25 return new DirectExchange(this.Exchange_Name);
 26 }
 27 
 28 @Bean
 29 public Binding bindingExchangeFirst(Queue queueFirst, DirectExchange directExchange){
 30 return BindingBuilder.bind(queueFirst).to(directExchange).with(Routing_Key_First);
 31 }
 32 
 33 @Bean
 34 public Queue queueDead(){
 35 return new Queue(this.Queue_Dead);
 36 }
 37 
 38 @Bean
 39 public DirectExchange directExchangeDead(){
 40 return new DirectExchange(this.Exchange_Dead);
 41 }
 42 
 43 @Bean
 44 public Binding bindingExchangeDead(Queue queueDead,DirectExchange directExchangeDead){
 45 return BindingBuilder.bind(queueDead).to(directExchangeDead).with(this.Routing_Key_Dead);
 46 }
 47 }
 48 
 49 @Configuration
 50 public class ConnectionConfig {
 51 @Value("${spring.rabbitmq.host}")
 52 public String host;
 53 
 54 @Value("${spring.rabbitmq.port}")
 55 public int port;
 56 
 57 @Value("${spring.rabbitmq.username}")
 58 public String username;
 59 
 60 @Value("${spring.rabbitmq.password}")
 61 public String password;
 62 
 63 @Value("${spring.rabbitmq.virtual-host}")
 64 public String virtualHost;
 65 
 66 @Bean
 67 public ConnectionFactory getConnectionFactory(){
 68 CachingConnectionFactory factory=new CachingConnectionFactory();
 69 factory.setHost(host);
 70 factory.setPort(port);
 71 factory.setUsername(username);
 72 factory.setPassword(password);
 73 factory.setVirtualHost(virtualHost);
 74 return factory;
 75 }
 76 }
 77 
 78 @Configuration
 79 public class DirectMessListener {
 80 @Autowired
 81 private ConnectionConfig connectionConfig;
 82 @Autowired
 83 private RabbitTemplate template;
 84 private int index=0,normalIndex=0,deadIndex=0; 
 85 
 86 @Bean
 87 public DirectMessageListenerContainer messageContainer(){
 88 DirectMessageListenerContainer container=new DirectMessageListenerContainer();
 89 container.setConnectionFactory(connectionConfig.getConnectionFactory());
 90 // 设置每个队列的 consumer 数量
 91 container.setConsumersPerQueue(4);
 92 // 设置每个 consumer 每次的接收的消息数量
 93 container.setPrefetchCount(10);
 94 // 使用MANUAL手动确认
 95 container.setAcknowledgeMode(AcknowledgeMode.MANUAL);
 96 // 监听队列
 97 container.addQueueNames(BindingConfig.Queue_First);
 98 container.addQueueNames(BindingConfig.Queue_Dead);
 99 container.setConsumerTagStrategy(queue -> "consumer"+(++index));
100 
101 container.setMessageListener(new ChannelAwareMessageListener(){
102 @Override
103 public void onMessage(Message message, com.rabbitmq.client.Channel channel) 
104 throws Exception {
105 MessageProperties prop=message.getMessageProperties();
106 if(prop.getReceivedRoutingKey().equals(BindingConfig.Routing_Key_First)){
107 System.out.println("This is a normal queue! "+(++normalIndex));
108 //把当前的队列转送到死信队列中
109 channel.basicReject(prop.getDeliveryTag(), false); 
110 }
111 if(prop.getReceivedRoutingKey().equals(BindingConfig.Routing_Key_Dead)){
112 System.out.println("This is a dead queue! "+(++deadIndex));
113 //模拟对死信队列处理
114 Thread.currentThread().sleep(5000);
115 .......
116 //处理完毕
117 channel.basicAck(prop.getDeliveryTag(), false);
118 }
119 
120 }
121 });
122 return container;
123 }
124 }