RabbitMQ工作模式介绍,交换机Exchange,不同工作模式的代码实现与Springboot整合

工作模式介绍
- 1.工作队列模式
- 2.发部订阅模式（广播模式）
- 3.路由全匹配模式
- 4. 路由模糊匹配模式
各种工作模式的代码实现- 原生客户端实现
- 1.工作队列模式
- - - 生产者
    - 消费者
- 2.发布订阅模式(广播模式)
- - 生产者
  - 消费者
- 3.路由全匹配模式
- - 生产者
  - 消费者
- 4.路由模糊配模式
- - 生产者
  - 消费者
各种工作模式的代码实现- SpringBoot整合与实现
- 一. 引入maven依赖
- 二.项目配置
- 三.队列配置
- - 1.工作队列模式
  - 2.发布订阅模式(广播模式)
  - 3.路由全匹配模式
  - 4.路由模糊匹配模式
- 四.生产者发送消息(各种模式不相同)
- 五、消费者队列监听
SpringMVC的整合
RabbitMQ高级知识
- 1.消息确认模式（consumer ack）
- 2.消息延时发送
- 3.消息超时消失TTL

工作模式介绍 1.工作队列模式

如官方的这个图和说明已经很清楚了-
Distributing tasks among workers：在多人中选择其中一个分发任务，描述的重点是多人中选择其中一个，如图中的C1与C2会选择其中一个进行处理，这也是分布式系统中的设计思想，目的是增加消费者的处理能力，也是工作中最常见的场景。

2.发部订阅模式（广播模式）

官方描述：Sending messages to many consumers at once-同时发送消息给多个消费者。如何理解呢?与工作队列模式图片中有什么不同与多了一个X：即Exchange，即交换机，消息不在直接发送给队列，而是发送给交换机，再由交换机发送给队列，至于交换机如何发送给队列，这就要看交换机的类型了，交换机的类型注意由如下3中：

Fanout，广播，发送给多有绑定该交换机的队列
Direct：定向，将消息发送给绑定了该交换机指定路由键（routing key）的队列
Toptic：通配符，将消息发送给绑定了该交换机且符合路由匹配规则（routing pattern）的队列

总结下发布订阅模式：

发布订阅模式就是声明交换机为Fanout 广播模式，能同时将消息发送给多个队列进行消费的模式，其实我更愿意叫做广播模式

3.路由全匹配模式

官方描述：Receiving messages selectively 选择性的接收消息，这里的接收消息可以理解为：交换机选择性的发送消息给指定的路由键（routing key）的绑定队列。这里的路由键是完全匹配，如果不完全匹配，是由一定的规则的，那就是下面的第四通配符匹配模式。
将第二种的发布订阅模式中的交换机类型定义为Direct，且不同的绑定队列定义不同的路由key，既是路由模式。（只要是理解了第二种，就应该能很好的理解第三种）。

4. 路由模糊匹配模式

理解了第二种、第三种模式，那这种模式就跟简单了，官方说明：Receiving messages based on a pattern – 通过匹配规则接收消息，即交换机通过路由键（routing key）的匹配规则将消息发送给对应的绑定队列。
这种模式的交换机类型定义为Toptic类型
路由匹配规则(routing pattern)如下:

：星号，能匹配一个字符，如图片：get.orange.now就能匹配.orange.;i.like.rabbit就能匹配.*.rabbit
#：井号，能匹配一道多个字符，如：lazy.或lazy.man.rabbit能匹配lazy.#

各种工作模式的代码实现- 原生客户端实现

准备：
创建公用的连接池方法

public class RabbitFactoryUtils {

    public static Connection newConnection() throws IOException, TimeoutException {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("192.168.1.13");
        factory.setPort(5672);
        factory.setUsername("clj");
        factory.setPassword("clj");
        factory.setVirtualHost("/clj");
        Connection connection = factory.newConnection();
        return connection;
    }
}

1.工作队列模式生产者

public class WorkQueuePublisher {

    public static void main(String[] args) throws IOException, TimeoutException {
        Connection connection = RabbitFactoryUtils.newConnection();
        Channel channel = connection.createChannel();
        //声明队列名
        //queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete, Map arguments)
        
        channel.queueDeclare("work_queue",false,false,false,null);
        //    public void basicPublish(String exchange, String routingKey, BasicProperties props, byte[] body) throws IOException
        channel.basicPublish("","work_queue",null,"工作队列模式work queue".getBytes());
        channel.close();
        connection.close();
        System.out.println("发送完毕");
    }
}

消费者

public class WorkQueueConsumer extends DefaultConsumer{

    public WorkQueueConsumer(Channel channel){
        super(channel);
    }

    
    @Override
    public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
        System.out.println("接收到数据"+consumerTag+"："+new String(body));
        System.out.println("deliveryTag"+envelope.getDeliveryTag());
        
        try {
            Thread.sleep(3000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        getChannel().basicAck(envelope.getDeliveryTag(),false);
    }

    public static void main(String[] args) throws IOException, TimeoutException {
        Connection connection = RabbitFactoryUtils.newConnection();
        Channel channel = connection.createChannel();
        
        channel.queueDeclare("work_queue",false,false,false,null);

        DefaultConsumer consumer = new WorkQueueConsumer(channel);

        
        channel.basicConsume("work_queue",false,consumer);

        //监听程序，不需要关闭资源

    }
}

2.发布订阅模式(广播模式) 生产者

public class FanoutPublisher {

    public static void main(String[] args) throws IOException, TimeoutException {
        Connection connection = RabbitFactoryUtils.newConnection();
        Channel channel = connection.createChannel();
        //声明队列名 队列名，
        
        channel.queueDeclare("pub_sub_queue1",true,false,false,null);
        channel.queueDeclare("pub_sub_queue2",true,false,false,null);
        //声明交换机器
        channel.exchangeDeclare("fanout_exchange", BuiltinExchangeType.FANOUT);

        //    public BindOk exchangeBind(String destination, String source, String routingKey) throws IOException {
        
        //    (String queue, String exchange, String routingKey)
        channel.queueBind("pub_sub_queue1","fanout_exchange","");
        channel.queueBind("pub_sub_queue2","fanout_exchange","");

        //String exchange, String routingKey, BasicProperties props, byte[] body
        channel.basicPublish("fanout_exchange","",null,"pub_sub_queue".getBytes());
        channel.close();
        connection.close();
        System.out.println("发送完毕");
    }
}

消费者

public class FanoutConsumer {

    public static void main(String[] args) throws IOException, TimeoutException {

        Connection connection = RabbitFactoryUtils.newConnection();
        Channel channel = connection.createChannel();
        //声明队列名 队列名，
        
        channel.queueDeclare("pub_sub_queue1",true,false,false,null);
        channel.queueDeclare("pub_sub_queue2",true,false,false,null);
        //声明交换机器
        channel.exchangeDeclare("fanout_exchange", BuiltinExchangeType.FANOUT);


        //    (String queue, String exchange, String routingKey)
        channel.queueBind("pub_sub_queue1","fanout_exchange","");
        channel.queueBind("pub_sub_queue2","fanout_exchange","");

        channel.basicConsume("pub_sub_queue1",false,new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("pub_sub_queue1-接收到数据"+consumerTag+"："+new String(body));
                getChannel().basicAck(envelope.getDeliveryTag(),true);
            }
        });


        channel.basicConsume("pub_sub_queue2",false,new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("pub_sub_queue2-接收到数据"+consumerTag+"："+new String(body));
                getChannel().basicAck(envelope.getDeliveryTag(),true);
            }
        });


    }
}

3.路由全匹配模式生产者

public class DirectPublisher {

    public static void main(String[] args) throws IOException, TimeoutException {
        Connection connection = RabbitFactoryUtils.newConnection();
        Channel channel = connection.createChannel();
        //声明队列名 队列名，
        //queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete, Map arguments)
        
        String exchangeName = "routing_key_exchange";
        String queue1 = "routing_key_queue1";
        String queue2 = "routing_key_queue2";

        String routing1 = "routing_key1";
        String routing2 = "routing_key2";
        channel.queueDeclare(queue1,true,false,false,null);
        channel.queueDeclare(queue2,true,false,false,null);
        //声明交换机器
        channel.exchangeDeclare(exchangeName, BuiltinExchangeType.DIRECT);

        
        channel.queueBind(queue1,exchangeName,routing1);
        channel.queueBind(queue2,exchangeName,routing2);

        //String exchange, String routingKey, BasicProperties props, byte[] body
        channel.basicPublish(exchangeName,routing1,null,(exchangeName+routing1+" message").getBytes());
        channel.basicPublish(exchangeName,routing2,null,(exchangeName+routing2+" message").getBytes());

        channel.close();
        connection.close();
        System.out.println("发送完毕");
    }
}

消费者

public class DirectConsumer {

    public static void main(String[] args) throws IOException, TimeoutException {
        Connection connection = RabbitFactoryUtils.newConnection();
        Channel channel = connection.createChannel();
        //声明队列名 队列名，
        
        String exchangeName = "routing_key_exchange";
        String queue1 = "routing_key_queue1";
        String queue2 = "routing_key_queue2";

        String routing1 = "routing_key1";
        String routing2 = "routing_key2";
        channel.queueDeclare(queue1,true,false,false,null);
        channel.queueDeclare(queue2,true,false,false,null);
        //声明交换机器
        channel.exchangeDeclare(exchangeName, BuiltinExchangeType.DIRECT);

        channel.queueBind(queue1,exchangeName,routing1);
        channel.queueBind(queue2,exchangeName,routing2);

        //String exchange, String routingKey, BasicProperties props, byte[] body
//        channel.basicPublish(exchangeName,routing1,null,(exchangeName+routing1+" message").getBytes());
//        channel.basicPublish(exchangeName,routing2,null,(exchangeName+routing2+" message").getBytes());


        channel.basicConsume(queue1,false,new DefaultConsumer(channel){

            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("queue1-接收到数据"+consumerTag+"："+new String(body));
                getChannel().basicAck(envelope.getDeliveryTag(),true);
            }
        });

        channel.basicConsume(queue2,false,new DefaultConsumer(channel){

            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("queue2-接收到数据"+consumerTag+"："+new String(body));
                getChannel().basicAck(envelope.getDeliveryTag(),true);
            }
        });
    }
}

4.路由模糊配模式生产者

public class TopicPublisher {

    public static void main(String[] args) throws IOException, TimeoutException {
        Connection connection = RabbitFactoryUtils.newConnection();
        Channel channel = connection.createChannel();
        //声明队列名 队列名，
        //queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete, Map arguments)
        
        String exchangeName = "topic_exchange";
        String queue1 = "topic_queue1";
        String queue2 = "topic_queue2";

        String routing1 = "routing_key1.*";
        String routing2 = "routing_key2.#";
        channel.queueDeclare(queue1,true,false,false,null);
        channel.queueDeclare(queue2,true,false,false,null);
        //声明交换机器
        channel.exchangeDeclare(exchangeName, BuiltinExchangeType.TOPIC);

        //    public BindOk exchangeBind(String destination, String source, String routingKey) throws IOException {
        
        //    (String queue, String exchange, String routingKey)
        channel.queueBind(queue1,exchangeName,routing1);
        channel.queueBind(queue2,exchangeName,routing2);

        //String exchange, String routingKey, BasicProperties props, byte[] body
        channel.basicPublish(exchangeName,"routing_key1.one",null,(exchangeName+routing1+" message").getBytes());
        channel.basicPublish(exchangeName,"routing_key2.two.test",null,(exchangeName+routing2+" message").getBytes());

        channel.close();
        connection.close();
        System.out.println("发送完毕");
    }

消费者

public class TopicConsumer {

    public static void main(String[] args) throws IOException, TimeoutException {
        Connection connection = RabbitFactoryUtils.newConnection();
        Channel channel = connection.createChannel();
        //声明队列名 队列名，
        //queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete, Map arguments)
        
        String exchangeName = "topic_exchange";
        String queue1 = "topic_queue1";
        String queue2 = "topic_queue2";

        String routing1 = "routing_key1.*";
        String routing2 = "routing_key2.#";
        channel.queueDeclare(queue1,true,false,false,null);
        channel.queueDeclare(queue2,true,false,false,null);
        //声明交换机器
        channel.exchangeDeclare(exchangeName, BuiltinExchangeType.TOPIC);

        //    (String queue, String exchange, String routingKey)
        channel.queueBind(queue1,exchangeName,routing1);
        channel.queueBind(queue2,exchangeName,routing2);

        //String exchange, String routingKey, BasicProperties props, byte[] body
        channel.basicPublish(exchangeName,"routing_key1.one",null,(exchangeName+routing1+" message").getBytes());
        channel.basicPublish(exchangeName,"routing_key2.two.test",null,(exchangeName+routing2+" message").getBytes());

        channel.basicConsume(queue1,false,new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println(queue1+"接收到数据"+consumerTag+"："+new String(body));
                getChannel().basicAck(envelope.getDeliveryTag(),true);            }
        });

        channel.basicConsume(queue2,false,new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println(queue2+"接收到数据"+consumerTag+"："+new String(body));
                getChannel().basicAck(envelope.getDeliveryTag(),true);            }
        });


    }
}

各种工作模式的代码实现- SpringBoot整合与实现一. 引入maven依赖

 
        
            org.springframework.boot
            spring-boot-starter-amqp

二.项目配置

spring:
  application:
    name: rabbitmq-test
  rabbitmq:
    virtual-host: /clj
    host: "192.168.1.13"
    port: 5672
    username: "clj"
    password: "clj"

三.队列配置 1.工作队列模式

@Configuration
public class WorkQueueRabbitConfig {

    public static final String WORK_QUEUE = "work_queue";


    @Bean(WORK_QUEUE)
    public Queue workQueue(){
        //仅声明队列即可，无需声明交换机
        return QueueBuilder.nonDurable(WORK_QUEUE).build();
    }
}

2.发布订阅模式(广播模式)

@Configuration
public class FanoutRabbitConfig {

    public static final String FANOUT_QUEUE1 = "sp_pub_sub_queue1";
    public static final String FANOUT_QUEUE2 = "sp_pub_sub_queue2";

    public static final String FANOUT_EXCHANGE = "sp_fanout_exchange";


    
    @Bean(FANOUT_QUEUE1)
    public Queue fanoutQueue1(){
        return QueueBuilder.nonDurable(FANOUT_QUEUE1).build();
    }

    @Bean(FANOUT_QUEUE2)
    public Queue fanoutQueue2(){
        return QueueBuilder.nonDurable(FANOUT_QUEUE2).build();
    }

    @Bean(FANOUT_EXCHANGE)
    public Exchange fanoutExchange(){
        return ExchangeBuilder.fanoutExchange(FANOUT_EXCHANGE).build();
    }

    
    @Bean
    public Binding bindingQueue1(@Qualifier(FANOUT_EXCHANGE) Exchange exchange, @Qualifier(FANOUT_QUEUE1) Queue queue){
        //默认路由key为空，exchange模式为Fanout时，即使设置routing key可是无效的
        return BindingBuilder.bind(queue).to(exchange).with("").noargs();
    }

    @Bean
    public Binding bindingQueue2(@Qualifier(FANOUT_EXCHANGE) Exchange exchange, @Qualifier(FANOUT_QUEUE2) Queue queue){
        return BindingBuilder.bind(queue).to(exchange).with("").noargs();
    }

}

3.路由全匹配模式

@Configuration
public class DirectRabbitConfig {


    public static final String DIRECT_QUEUE1 = "sp_direct_queue1";
    public static final String DIRECT_QUEUE2 = "sp_direct_queue2";

    public static final String DIRECT_EXCHANGE = "sp_direct_exchange";

    public static final String DIRECT_ROUTING_KEY1 = "sp_routing_key1";
    public static final String DIRECT_ROUTING_KEY2 = "sp_routing_key2";


    
    @Bean(DIRECT_QUEUE1)
    public Queue fanoutQueue1(){
        return QueueBuilder.nonDurable(DIRECT_QUEUE1).build();
    }

    @Bean(DIRECT_QUEUE2)
    public Queue fanoutQueue2(){
        return QueueBuilder.nonDurable(DIRECT_QUEUE2).build();
    }

    @Bean(DIRECT_EXCHANGE)
    public Exchange fanoutExchange(){
        return ExchangeBuilder.directExchange(DIRECT_EXCHANGE).build();
    }

    
    @Bean
    public Binding bindingQueue1(@Qualifier(DIRECT_EXCHANGE) Exchange exchange, @Qualifier(DIRECT_QUEUE1) Queue queue){
        //默认路由key为空，exchange模式为Fanout时，这里需要绑定路由key
        return BindingBuilder.bind(queue).to(exchange).with(DIRECT_ROUTING_KEY1).noargs();
    }

    @Bean
    public Binding bindingQueue2(@Qualifier(DIRECT_EXCHANGE) Exchange exchange, @Qualifier(DIRECT_QUEUE2) Queue queue){
        return BindingBuilder.bind(queue).to(exchange).with(DIRECT_ROUTING_KEY2).noargs();
    }

}

4.路由模糊匹配模式

@Configuration
public class TopicRabbitConfig {

    public static final String TOPIC_QUEUE1 = "sp_topic_queue1";
    public static final String TOPIC_QUEUE2 = "sp_topic_queue2";

    public static final String TOPIC_EXCHANGE = "sp_topic_exchange";

    public static final String TOPIC_ROUTING_KEY1 = "sp_routing_key1.*";
    public static final String TOPIC_ROUTING_KEY2 = "sp_routing_key2.#";

    
    @Bean(TOPIC_QUEUE1)
    public Queue fanoutQueue1(){
        return QueueBuilder.nonDurable(TopicRabbitConfig.TOPIC_QUEUE1).build();
    }

    @Bean(TOPIC_QUEUE2)
    public Queue fanoutQueue2(){
        return QueueBuilder.nonDurable(TOPIC_QUEUE2).build();
    }

    @Bean(TOPIC_EXCHANGE)
    public Exchange fanoutExchange(){
        return ExchangeBuilder.topicExchange(TopicRabbitConfig.TOPIC_EXCHANGE).build();
    }

    
    @Bean
    public Binding bindingTopicQueue1(@Qualifier(TOPIC_EXCHANGE) Exchange exchange, @Qualifier(TOPIC_QUEUE1) Queue queue){
        //默认路由key为空，exchange模式为Fanout时，这里需要绑定路由key
        return BindingBuilder.bind(queue).to(exchange).with(TOPIC_ROUTING_KEY1).noargs();
    }

    @Bean
    public Binding bindingTopicQueue2(@Qualifier(TOPIC_EXCHANGE) Exchange exchange, @Qualifier(TOPIC_QUEUE2) Queue queue){
        return BindingBuilder.bind(queue).to(exchange).with(TOPIC_ROUTING_KEY2).noargs();
    }

}

四.生产者发送消息(各种模式不相同)

@RunWith(SpringRunner.class)
@SpringBootTest
public class PublisherTest {

    @Resource
    private AmqpTemplate rabbitTemplate;

    
    @Test
    public void testWorkQueue(){
        for (int i = 1; i < 101; i++) {
            rabbitTemplate.convertAndSend(WorkQueueRabbitConfig.WORK_QUEUE,i+"spring boot work queue !");
        }
    }

    
    @Test
    public void testFanout(){
        for (int i = 1; i < 101; i++) {
            
            rabbitTemplate.convertAndSend(FanoutRabbitConfig.FANOUT_EXCHANGE,"",i+"spring boot fanout queue !");
        }
    }


    
    @Test
    public void testDirect(){
        for (int i = 1; i < 101; i++) {
            
            rabbitTemplate.convertAndSend(DirectRabbitConfig.DIRECT_EXCHANGE,DirectRabbitConfig.DIRECT_ROUTING_KEY1,i+"spring boot direct queue ,routing key 1 !");
            rabbitTemplate.convertAndSend(DirectRabbitConfig.DIRECT_EXCHANGE,DirectRabbitConfig.DIRECT_ROUTING_KEY2,i+"spring boot direct queue ,routing key 2 !");
        }
    }

    
    @Test
    public void testTopic(){
        for (int i = 1; i < 101; i++) {
            
            rabbitTemplate.convertAndSend(TopicRabbitConfig.TOPIC_EXCHANGE,"sp_routing_key1.info",i+"spring boot topic queue ,routing key 1 info !");
            rabbitTemplate.convertAndSend(TopicRabbitConfig.TOPIC_EXCHANGE,"sp_routing_key2.error.task",i+"spring boot topic queue ,routing key 2 error.task !");
        }
    }
}

五、消费者队列监听

各种工作模式的监听方式一样，只需要监听具体队列名即可

@Component
public class WorkQueueListener {

    @RabbitListener(queues = WorkQueueRabbitConfig.WORK_QUEUE)
    public void ListenerWorkQueue(Message message){
        System.out.println("ListenerWorkQueue 收到监听:"+new String(message.getBody()));
    }

    @RabbitListener(queues = FanoutRabbitConfig.FANOUT_QUEUE1)
    public void ListenerFanoutQueue1(Message message){

        System.out.println("ListenerFanoutQueue1 收到监听:"+new String(message.getBody()));
    }

    @RabbitListener(queues = FanoutRabbitConfig.FANOUT_QUEUE2)
    public void ListenerFanoutQueue2(Message message){

        System.out.println("ListenerFanoutQueue2 收到监听:"+new String(message.getBody()));
    }

    @RabbitListener(queues = DirectRabbitConfig.DIRECT_QUEUE1)
    public void ListenerDirectQueue1(Message message){

        System.out.println("ListenerDirectQueue1 收到监听:"+new String(message.getBody()));
    }

    @RabbitListener(queues = DirectRabbitConfig.DIRECT_QUEUE2)
    public void ListenerDirectQueue2(Message message){

        System.out.println("ListenerDirectQueue2 收到监听:"+new String(message.getBody()));
    }

    @RabbitListener(queues = TopicRabbitConfig.TOPIC_QUEUE1)
    public void ListenerTopicQueue1(Message message){

        System.out.println("ListenerTopicQueue1 收到监听:"+new String(message.getBody()));
    }

    @RabbitListener(queues = TopicRabbitConfig.TOPIC_QUEUE2)
    public void ListenerTopicQueue2(Message message){

        System.out.println("ListenerTopicQueue2 收到监听:"+new String(message.getBody()));
    }
}

SpringMVC的整合

见文章《springrabbitmq中使用@RabbitListener实现监听消息-传统spring-springmvc项目整合rabbitmq》

RabbitMQ高级知识 1.消息确认模式（consumer ack）

ack，即acknowledge，确认，消息确认模式有三种

none ：自动确认
manual：手动确认
auto ：根据异常情况确认
channel.basicAck()可以确认消息，channel.basicNack()

2.消息延时发送 3.消息超时消失TTL

RabbitMQ工作模式介绍,交换机Exchange,不同工作模式的代码实现与Springboot整合

Java相关栏目本月热门文章