概念

rabbitmq概念比较多，这里说几个重要的

routing key: 消息体的一部分 发送消息时指定，用于队列和交换机绑定关系进行匹配。

exchange 交换机:rabbitmq发送消息直接发送到交换机，由交换机发送消息给队列。

queue：用于存储消息，交换机发送的消息会进入这里面，在创建的时候可以指定消息持久化，设置durable属性为TRUE。

bingKey 用于绑定交换机和队列的关系，在项目初始化的时候声明，bindkey决定了通配符的消息进入哪些队列

通配符：支持 *和# 比如 a.b.c 就是三个单词，单词多少以点为界限确定 #可以匹配多个单词*只可以匹配一个单词

简单结构图如下：

发送方式

rabbitmq支持了点对点，通配符，广播模式等发送方式。

下面主要就这三点做一个简单概述,示例代码使用springboot

点对点

发送消息指定消息routing key 为 a.b ,只有接收端交换机和队列使用a.b绑定的时候才可以使用

配置类

package com.tom.config;import com.tom.message.Demo01Message;import org.springframework.amqp.core.*;import org.springframework.context.annotation.Bean;import org.springframework.context.annotation.Configuration;@Configurationpublic class RabbitConfig {    /**     * Direct Exchange 示例的配置类     */    public static class DirectExchangeDemoConfiguration { // 创建 Queue //点对点发布 @Bean public Queue demo01Queue() {     return new Queue(Demo01Message.QUEUE, // Queue 名字      true, // durable: 是否持久化      false, // exclusive: 是否排它      false); // autoDelete: 是否自动删除 } // 创建 Direct Exchange @Bean public DirectExchange demo01Exchange() {     return new DirectExchange(Demo01Message.EXCHANGE,      true,  // durable: 是否持久化      false);  // exclusive: 是否排它 } // 创建 Binding // Exchange：Demo01Message.EXCHANGE // Routing key：Demo01Message.ROUTING_KEY // Queue：Demo01Message.QUEUE @Bean public Binding demo01Binding() {     return BindingBuilder.bind(demo01Queue()).to(demo01Exchange()).with(Demo01Message.ROUTING_KEY); }    }    }

消费者：

package com.tom.mq;// Demo01Consumer.javaimport com.tom.message.Demo01Message;import org.slf4j.Logger;import org.slf4j.LoggerFactory;import org.springframework.amqp.rabbit.annotation.RabbitHandler;import org.springframework.amqp.rabbit.annotation.RabbitListener;import org.springframework.stereotype.Component;@Component@RabbitListener(queues = Demo01Message.QUEUE,concurrency = "5")public class Demo01Consumer {    private Logger logger = LoggerFactory.getLogger(getClass());    @RabbitHandler    public void onMessage(Demo01Message message) { logger.info("[onMessage][线程编号:{} 消息内容：{}]", Thread.currentThread().getId(), message);    }}

发送端：

 rabbitTemplate.convertAndSend(Demo01Message.EXCHANGE, Demo01Message.ROUTING_KEY, message);

topic方式发送

配置类，在点对点的配置类的基础上加上内部类

/**     * 下面的配置为通配符发布 topic发布     */    public static  class LogExchangeConfiguration{ // 创建 Queue @Bean public Queue logQueue1() {     return new Queue(Constant.myQueue1, // Queue 名字      true, // durable: 是否持久化      false, // exclusive: 是否排它      false); // autoDelete: 是否自动删除 } @Bean public Queue logQueue2() {     return new Queue(Constant.myQueue2, // Queue 名字      true, // durable: 是否持久化      false, // exclusive: 是否排它      false); // autoDelete: 是否自动删除 } // 创建 Topic Exchange @Bean public TopicExchange logExchange() {     return new TopicExchange(Constant.myExchange,      true,  // durable: 是否持久化      false);  // exclusive: 是否排它 } // 创建 Binding //一个queue 多个 @Bean public Binding BindingLogQueue1() {     return BindingBuilder.bind(logQueue1()).to(logExchange()).with("a.b"); } @Bean public Binding bindingLogQueue2() {     return BindingBuilder.bind(logQueue2()).to(logExchange()).with("a.*"); }    }

消费者：

package com.tom.mq;// Demo01Consumer.javaimport com.tom.config.Constant;import com.tom.message.Demo01Message;import org.slf4j.Logger;import org.slf4j.LoggerFactory;import org.springframework.amqp.rabbit.annotation.RabbitHandler;import org.springframework.amqp.rabbit.annotation.RabbitListener;import org.springframework.stereotype.Component;@Component@RabbitListener(queues = Constant.myQueue1,concurrency = "5")public class Consumer1 {    private Logger logger = LoggerFactory.getLogger(getClass());    @RabbitHandler    public void onMessage( String message) { logger.info("myQueue1收到消息【{}】", message);    }}

package com.tom.mq;// Demo01Consumer.javaimport com.tom.config.Constant;import org.slf4j.Logger;import org.slf4j.LoggerFactory;import org.springframework.amqp.rabbit.annotation.RabbitHandler;import org.springframework.amqp.rabbit.annotation.RabbitListener;import org.springframework.stereotype.Component;@Component@RabbitListener(queues = Constant.myQueue2,concurrency = "5")public class Consumer2 {    private Logger logger = LoggerFactory.getLogger(getClass());    @RabbitHandler    public void onMessage( String message) { logger.info("myQueue2收到消息{}", message);    }}

发送端：

package com.tom;import com.tom.config.Constant;import org.junit.Test;import org.junit.runner.RunWith;import org.springframework.amqp.rabbit.core.RabbitTemplate;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.boot.test.context.SpringBootTest;import org.springframework.test.context.junit4.SpringRunner;@RunWith(SpringRunner.class)@SpringBootTest(classes = Application.class)public class TopicTest {    @Autowired    private RabbitTemplate rabbitTemplate;    @Test    public void test() { String  message="hello a--------------------->>>>>>>>>>>>>>>>>>>>>"; rabbitTemplate.convertAndSend(Constant.myExchange,"a.b",message); System.out.println("发送消息success！！！！");    }}

发送结果：

 发布模式和上面两种类似，这里暂时不做赘述。

开发总结

1.声明队列，交换机，建立交换机和队列的绑定关系

2.建立监听队列的监听器

3.发送消息指定消息的routingkey

4.接收消息处理业务

代码地址：

 https://github.com/weiwensi/cloud-parent.git

地址上项目的rabbitmq模块