|
| 1 | +package com.xncoding.pos.config; |
| 2 | + |
| 3 | +import org.slf4j.Logger; |
| 4 | +import org.slf4j.LoggerFactory; |
| 5 | +import org.springframework.amqp.core.*; |
| 6 | +import org.springframework.amqp.rabbit.core.RabbitTemplate; |
| 7 | +import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter; |
| 8 | +import org.springframework.beans.factory.annotation.Qualifier; |
| 9 | +import org.springframework.context.annotation.Bean; |
| 10 | +import org.springframework.context.annotation.Configuration; |
| 11 | + |
| 12 | +import javax.annotation.Resource; |
| 13 | + |
| 14 | +/** |
| 15 | + * RabbitConfig |
| 16 | + * |
| 17 | + * @author XiongNeng |
| 18 | + * @version 1.0 |
| 19 | + * @since 2018/3/1 |
| 20 | + */ |
| 21 | +@Configuration |
| 22 | +public class RabbitConfig { |
| 23 | + @Resource |
| 24 | + private RabbitTemplate rabbitTemplate; |
| 25 | + |
| 26 | + /** |
| 27 | + * 定制化amqp模版 可根据需要定制多个 |
| 28 | + * <p> |
| 29 | + * <p> |
| 30 | + * 此处为模版类定义 Jackson消息转换器 |
| 31 | + * ConfirmCallback接口用于实现消息发送到RabbitMQ交换器后接收ack回调 即消息发送到exchange ack |
| 32 | + * ReturnCallback接口用于实现消息发送到RabbitMQ 交换器，但无相应队列与交换器绑定时的回调 即消息发送不到任何一个队列中 ack |
| 33 | + * |
| 34 | + * @return the amqp template |
| 35 | + */ |
| 36 | + // @Primary |
| 37 | + @Bean |
| 38 | + public AmqpTemplate amqpTemplate() { |
| 39 | + Logger log = LoggerFactory.getLogger(RabbitTemplate.class); |
| 40 | + // 使用jackson 消息转换器 |
| 41 | + rabbitTemplate.setMessageConverter(new Jackson2JsonMessageConverter()); |
| 42 | + rabbitTemplate.setEncoding("UTF-8"); |
| 43 | + // 消息发送失败返回到队列中，yml需要配置 publisher-returns: true |
| 44 | + rabbitTemplate.setMandatory(true); |
| 45 | + rabbitTemplate.setReturnCallback((message, replyCode, replyText, exchange, routingKey) -> { |
| 46 | + String correlationId = message.getMessageProperties().getCorrelationIdString(); |
| 47 | + log.debug("消息：{} 发送失败, 应答码：{} 原因：{} 交换机: {} 路由键: {}", correlationId, replyCode, replyText, exchange, routingKey); |
| 48 | + }); |
| 49 | + // 消息确认，yml需要配置 publisher-confirms: true |
| 50 | + rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> { |
| 51 | + if (ack) { |
| 52 | + log.debug("消息发送到exchange成功,id: {}", correlationData.getId()); |
| 53 | + } else { |
| 54 | + log.debug("消息发送到exchange失败,原因: {}", cause); |
| 55 | + } |
| 56 | + }); |
| 57 | + return rabbitTemplate; |
| 58 | + } |
| 59 | + |
| 60 | + /* ----------------------------------------------------------------------------Direct exchange test--------------------------------------------------------------------------- */ |
| 61 | + |
| 62 | + /** |
| 63 | + * 声明Direct交换机 支持持久化. |
| 64 | + * |
| 65 | + * @return the exchange |
| 66 | + */ |
| 67 | + @Bean("directExchange") |
| 68 | + public Exchange directExchange() { |
| 69 | + return ExchangeBuilder.directExchange("DIRECT_EXCHANGE").durable(true).build(); |
| 70 | + } |
| 71 | + |
| 72 | + /** |
| 73 | + * 声明一个队列 支持持久化. |
| 74 | + * |
| 75 | + * @return the queue |
| 76 | + */ |
| 77 | + @Bean("directQueue") |
| 78 | + public Queue directQueue() { |
| 79 | + return QueueBuilder.durable("DIRECT_QUEUE").build(); |
| 80 | + } |
| 81 | + |
| 82 | + /** |
| 83 | + * 通过绑定键 将指定队列绑定到一个指定的交换机 . |
| 84 | + * |
| 85 | + * @param queue the queue |
| 86 | + * @param exchange the exchange |
| 87 | + * @return the binding |
| 88 | + */ |
| 89 | + @Bean |
| 90 | + public Binding directBinding(@Qualifier("directQueue") Queue queue, |
| 91 | + @Qualifier("directExchange") Exchange exchange) { |
| 92 | + return BindingBuilder.bind(queue).to(exchange).with("DIRECT_ROUTING_KEY").noargs(); |
| 93 | + } |
| 94 | + |
| 95 | + /* ----------------------------------------------------------------------------Fanout exchange test--------------------------------------------------------------------------- */ |
| 96 | + |
| 97 | + /** |
| 98 | + * 声明 fanout 交换机. |
| 99 | + * |
| 100 | + * @return the exchange |
| 101 | + */ |
| 102 | + @Bean("fanoutExchange") |
| 103 | + public FanoutExchange fanoutExchange() { |
| 104 | + return (FanoutExchange) ExchangeBuilder.fanoutExchange("FANOUT_EXCHANGE").durable(true).build(); |
| 105 | + } |
| 106 | + |
| 107 | + /** |
| 108 | + * Fanout queue A. |
| 109 | + * |
| 110 | + * @return the queue |
| 111 | + */ |
| 112 | + @Bean("fanoutQueueA") |
| 113 | + public Queue fanoutQueueA() { |
| 114 | + return QueueBuilder.durable("FANOUT_QUEUE_A").build(); |
| 115 | + } |
| 116 | + |
| 117 | + /** |
| 118 | + * Fanout queue B . |
| 119 | + * |
| 120 | + * @return the queue |
| 121 | + */ |
| 122 | + @Bean("fanoutQueueB") |
| 123 | + public Queue fanoutQueueB() { |
| 124 | + return QueueBuilder.durable("FANOUT_QUEUE_B").build(); |
| 125 | + } |
| 126 | + |
| 127 | + /** |
| 128 | + * 绑定队列A 到Fanout 交换机. |
| 129 | + * |
| 130 | + * @param queue the queue |
| 131 | + * @param fanoutExchange the fanout exchange |
| 132 | + * @return the binding |
| 133 | + */ |
| 134 | + @Bean |
| 135 | + public Binding bindingA(@Qualifier("fanoutQueueA") Queue queue, |
| 136 | + @Qualifier("fanoutExchange") FanoutExchange fanoutExchange) { |
| 137 | + return BindingBuilder.bind(queue).to(fanoutExchange); |
| 138 | + } |
| 139 | + |
| 140 | + /** |
| 141 | + * 绑定队列B 到Fanout 交换机. |
| 142 | + * |
| 143 | + * @param queue the queue |
| 144 | + * @param fanoutExchange the fanout exchange |
| 145 | + * @return the binding |
| 146 | + */ |
| 147 | + @Bean |
| 148 | + public Binding bindingB(@Qualifier("fanoutQueueB") Queue queue, |
| 149 | + @Qualifier("fanoutExchange") FanoutExchange fanoutExchange) { |
| 150 | + return BindingBuilder.bind(queue).to(fanoutExchange); |
| 151 | + } |
| 152 | +} |
0 commit comments