springboot2整合rabbitmq,並介紹3種常見exchange
阿新 • • 發佈:2018-11-03
1.新增Maven依賴
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
2.application.properties配置
spring.rabbitmq.virtual-host=test spring.rabbitmq.host=182.92.234.232 spring.rabbitmq.port=5672 spring.rabbitmq.username=admin spring.rabbitmq.password=123456
3.FanoutExchange
@Configuration public class RabbitmqFanoutConfig { public static final String FANOUT_QUEUE_A = "fanout_queue_a"; public static final String FANOUT_QUEUE_B = "fanout_queue_b"; public static final String FANOUT_QUEUE_C = "fanout_queue_c"; public static final String FANOUT_EXCHANGE = "fanout_exchange"; @Bean public Queue fanoutQueueA() { return new Queue(FANOUT_QUEUE_A); } @Bean public Queue fanoutQueueB() { return new Queue(FANOUT_QUEUE_B); } @Bean public Queue fanoutQueueC() { return new Queue(FANOUT_QUEUE_C); } @Bean public FanoutExchange fanoutExchange() { return new FanoutExchange(FANOUT_EXCHANGE); } @Bean public Binding queueABindingExchange(Queue fanoutQueueA, FanoutExchange fanoutExchange){ return BindingBuilder.bind(fanoutQueueA).to(fanoutExchange); } @Bean public Binding queueBBindingExchange(Queue fanoutQueueB, FanoutExchange fanoutExchange){ return BindingBuilder.bind(fanoutQueueB).to(fanoutExchange); } @Bean public Binding queueCBindingExchange(Queue fanoutQueueC, FanoutExchange fanoutExchange){ return BindingBuilder.bind(fanoutQueueC).to(fanoutExchange); } } @Component public class FanoutProducer { private static final Logger logger = LoggerFactory.getLogger(FanoutProducer.class); @Autowired private AmqpTemplate amqpTemplate; public void fanoutSendMessage(Message message) { logger.info("[===========> FanoutExchange Send Message] body: " + message.toString()); amqpTemplate.convertAndSend(RabbitmqFanoutConfig.FANOUT_EXCHANGE, "", message); } } @Component public class FanoutConsumers { private static final Logger logger = LoggerFactory.getLogger(FanoutConsumers.class); @RabbitListener(queues = RabbitmqFanoutConfig.FANOUT_QUEUE_A) public void fanoutQueueA(Message message) { logger.info("[===========> Message From FanoutQueueA] body: " + message.toString()); } @RabbitListener(queues = RabbitmqFanoutConfig.FANOUT_QUEUE_B) public void fanoutQueueB(Message message) { logger.info("[===========> Message From FanoutQueueB] body: " + message.toString()); } @RabbitListener(queues = RabbitmqFanoutConfig.FANOUT_QUEUE_C) public void fanoutQueueC(Message message) { logger.info("[===========> Message From FanoutQueueC] body: " + message.toString()); } } @Autowired private FanoutProducer fanoutProducer; @Test public void testFanoutExchange() { Message message = new Message(); message.setTitle("fanoutExchange"); message.setContent("send message to fanoutExchange"); fanoutProducer.fanoutSendMessage(message); } fanout可以理解為廣播模式或者訂閱模式, 給fanout交換機發送訊息, 綁定了這個交換機的所有佇列都收到這個訊息.
4.TopicExchange
@Configuration public class RabbitmqTopicConfig { public static final String TOPIC_QUEUE_A = "topic.a"; public static final String TOPIC_QUEUE_B = "topic.b"; public static final String TOPIC_QUEUE_OTHER = "topic.other"; public static final String TOPIC_ROUTING_KEY_A = "topic.a"; public static final String TOPIC_ROUTING_KEY_B = "topic.*"; public static final String TOPIC_ROUTING_KEY_C = "topic.#"; public static final String TOPIC_EXCHANGE = "topic_exchange"; @Bean public Queue topicQueueA() { return new Queue(TOPIC_QUEUE_A); } @Bean public Queue topicQueueB() { return new Queue(TOPIC_QUEUE_B); } @Bean public Queue topicQueueOther() { return new Queue(TOPIC_QUEUE_OTHER); } @Bean public TopicExchange topicExchange() { return new TopicExchange(TOPIC_EXCHANGE); } @Bean public Binding topicQueueABindingTopicExchange(Queue topicQueueA, TopicExchange topicExchange){ return BindingBuilder.bind(topicQueueA).to(topicExchange).with(TOPIC_ROUTING_KEY_A); } @Bean public Binding topicQueueBBindingTopicExchange(Queue topicQueueB, TopicExchange topicExchange){ return BindingBuilder.bind(topicQueueB).to(topicExchange).with(TOPIC_ROUTING_KEY_B); } @Bean public Binding topicQueueCBindingTopicExchange(Queue topicQueueOther, TopicExchange topicExchange){ return BindingBuilder.bind(topicQueueOther).to(topicExchange).with(TOPIC_ROUTING_KEY_C); } } @Component public class TopicProducer { private static final Logger logger = LoggerFactory.getLogger(TopicProducer.class); @Autowired private AmqpTemplate amqpTemplate; public void topicSendMessage(Message message, String routingKey) { logger.info("[===========> TopicExchange Send Message] body: " + message.toString()); amqpTemplate.convertAndSend(RabbitmqTopicConfig.TOPIC_EXCHANGE, routingKey, message); } } @Component public class TopicConsumers { private static final Logger logger = LoggerFactory.getLogger(TopicConsumers.class); @RabbitListener(queues = RabbitmqTopicConfig.TOPIC_QUEUE_A) public void topicQueueA(Message message) { logger.info("[===========> Message From TopicQueueA] body: " + message.toString()); } @RabbitListener(queues = RabbitmqTopicConfig.TOPIC_QUEUE_B) public void topicQueueB(Message message) { logger.info("[===========> Message From TopicQueueB] body: " + message.toString()); } @RabbitListener(queues = RabbitmqTopicConfig.TOPIC_QUEUE_OTHER) public void topicQueueOther(Message message) { logger.info("[===========> Message From TopicQueueOther] body: " + message.toString()); } } @Autowired private TopicProducer topicProducer; @Test public void testTopicExchange001() { // 消費者A B Other都能收到 String routingKey = "topic.a"; Message message = new Message(); message.setTitle("topicExchange"); message.setContent("send message to topicExchange"); topicProducer.topicSendMessage(message, routingKey); } @Test public void testTopicExchange002() { // 消費者B Other都能收到 String routingKey = "topic.word001"; Message message = new Message(); message.setTitle("topicExchange"); message.setContent("send message to topicExchange"); topicProducer.topicSendMessage(message, routingKey); } @Test public void testTopicExchange003() { // 消費者Other能收到 String routingKey = "topic.word001.word002"; Message message = new Message(); message.setTitle("topicExchange"); message.setContent("send message to topicExchange"); topicProducer.topicSendMessage(message, routingKey); } topic是最靈活的方式, 可以根據routingKey自由的傳送訊息到繫結的不同佇列. (#表示匹配多個詞, *表示匹配一個詞)
5.DirectExchange
@Component
public class RabbitmqDirectConfig {
public static final String DIRECT_QUEUE_A = "direct_queue_a";
public static final String DIRECT_QUEUE_B = "direct_queue_b";
public static final String DIRECT_EXCHANGE = "direct_exchange";
public static final String DIRECT_ROUTING_KEY_A = "direct.a";
public static final String DIRECT_ROUTING_KEY_B = "direct.#";
@Bean
public Queue directQueueA() {
return new Queue(DIRECT_QUEUE_A);
}
@Bean
public Queue directQueueB() {
return new Queue(DIRECT_QUEUE_B);
}
@Bean
public DirectExchange directExchange() {
return new DirectExchange(DIRECT_EXCHANGE);
}
@Bean
public Binding directBinding001(Queue directQueueA, DirectExchange directExchange) {
return BindingBuilder.bind(directQueueA).to(directExchange).with(DIRECT_ROUTING_KEY_A);
}
@Bean
public Binding directBinding002(Queue directQueueB, DirectExchange directExchange) {
return BindingBuilder.bind(directQueueB).to(directExchange).with(DIRECT_ROUTING_KEY_B);
}
}
@Component
public class DirectProducer {
private static final Logger logger = LoggerFactory.getLogger(DirectProducer.class);
@Autowired
private AmqpTemplate amqpTemplate;
public void directSendMessage(Message message, String routingKey) {
logger.info("[===========> DirectExchange Send Message] body: " + message.toString());
amqpTemplate.convertAndSend(RabbitmqDirectConfig.DIRECT_EXCHANGE, routingKey, message);
}
}
@Component
public class DirectConsumer {
private static final Logger logger = LoggerFactory.getLogger(DirectConsumer.class);
@RabbitListener(queues = RabbitmqDirectConfig.DIRECT_QUEUE_A)
public void directQueueA(Message message) {
logger.info("[===========> Message From DirectQueueA] body: " + message.toString());
}
@RabbitListener(queues = RabbitmqDirectConfig.DIRECT_QUEUE_B)
public void directQueueB(Message message) {
logger.info("[===========> Message From DirectQueueB] body: " + message.toString());
}
}
@Autowired
private DirectProducer directProducer;
@Test
public void testDirectExchange001() {
// 消費者A能收到
String routingKey = "direct.a";
Message message = new Message();
message.setTitle("directExchange");
message.setContent("send message to directExchange");
directProducer.directSendMessage(message, routingKey);
}
@Test
public void testDirectExchange002() {
// 誰都收不到
String routingKey = "direct.b";
Message message = new Message();
message.setTitle("directExchange");
message.setContent("send message to directExchange");
directProducer.directSendMessage(message, routingKey);
}
direct是直接匹配, 沒有什麼可說的
6.總結
相對來說還是比較簡單的, 更深入的文章會在後面
原始碼 https://gitee.com/jsjack_wang/springboot-demo dev-rabbitmq分支