routing路由模式

一：介绍

1.模式

　　

2.应用场景

　　如果exchangge与队列中的key相同，消息就发送过去。

　　这个就是需要将交换机与队列增加key。

3.路由类型

　　上节课的订阅模式中的路由类型是Fanout。

　　这篇文章的路由类型是Direct。

二：程序

1.生产者

package com.mq.routing;

import com.mq.utils.ConnectionUtil;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

public class RoutingSend {
    private static final String EXCHANGE_NAME="test_exchange_direct";
    private static final String QUEUE_NAME="test_queue_direct_1";
    public static void main(String[] args)throws Exception{
        Connection connection= ConnectionUtil.getConnection();
        Channel channel=connection.createChannel();
        channel.exchangeDeclare(EXCHANGE_NAME,"direct");
        String msg="hello routing";
        //重点是第二个参数routingKey
        String routingKey="info";
        channel.basicPublish(EXCHANGE_NAME,routingKey,null,msg.getBytes());
        System.out.println("msg send:"+msg);
        channel.close();
        connection.close();
    }
}

2.消费者一

package com.mq.routing;

import com.mq.utils.ConnectionUtil;
import com.rabbitmq.client.*;

import java.io.IOException;

public class RoutingReceive1 {
    private static final String EXCHANGE_NAME="test_exchange_direct";
    private static final String QUEUE_NAME="test_queue_direct_1";
    public static void main(String[] args)throws Exception{
        Connection connection= ConnectionUtil.getConnection();
        final Channel channel=connection.createChannel();
        channel.queueDeclare(QUEUE_NAME,false,false,false,null);
        channel.queueBind(QUEUE_NAME,EXCHANGE_NAME,"error");
        channel.basicQos(1);
        Consumer consumer=new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                String msg = new String(body, "utf-8");
                System.out.println("[1] receive:" + msg);
                try {
                    Thread.sleep(200);
                }catch (Exception e){
                    e.printStackTrace();
                }finally {
                    System.out.println("[done]");
                    channel.basicAck(envelope.getDeliveryTag(),false);
                }
            }
        };
        boolean autoAck=false;
        channel.basicConsume(QUEUE_NAME,autoAck,consumer);
    }
}

3.消费者二

package com.mq.routing;

import com.mq.utils.ConnectionUtil;
import com.rabbitmq.client.*;

import java.io.IOException;

public class RoutingReceive2 {
    private static final String EXCHANGE_NAME="test_exchange_direct";
    private static final String QUEUE_NAME="test_queue_direct_2";
    public static void main(String[] args)throws Exception{
        Connection connection= ConnectionUtil.getConnection();
        final Channel channel=connection.createChannel();
        channel.queueDeclare(QUEUE_NAME,false,false,false,null);
        channel.queueBind(QUEUE_NAME,EXCHANGE_NAME,"error");
        channel.queueBind(QUEUE_NAME,EXCHANGE_NAME,"info");
        channel.queueBind(QUEUE_NAME,EXCHANGE_NAME,"waring");
        channel.basicQos(1);
        Consumer consumer=new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                String msg = new String(body, "utf-8");
                System.out.println("[1] receive:" + msg);
                try {
                    Thread.sleep(200);
                }catch (Exception e){
                    e.printStackTrace();
                }finally {
                    System.out.println("[done]");
                    channel.basicAck(envelope.getDeliveryTag(),false);
                }
            }
        };
        boolean autoAck=false;
        channel.basicConsume(QUEUE_NAME,autoAck,consumer);
    }
}

4.现象

　　send：

　　

　　receive1：

　　

　　receive2：