Routing 路由模式

        Routing 模式又称路由模式，该种模式除了要绑定交换机外，发消息的时候还要制定routing key，即路由key，队列通过通道绑定交换机的时候，需要指定自己的routing key，这样，生产端发送消息的时候也会指定routing key，通过routing key就可以把相应的消息发送到绑定相应routing key的队列中去。

缺失图片

特点：

1、每个消费者监听自己的队列，并且设置routingkey；
2、生产者将消息发给交换机，由交换机根据routingkey来转发消息到指定的队列；

应用场景：用户通知，当用户充值成功或转账完成系统通知用户，通知方式有短信、邮件多种方法；

routing路由模式之生产者

package com.zzg.rabbitmq.routing;
 
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
 
import java.io.IOException;
import java.util.concurrent.TimeoutException;
 
/**
 * 生产者核心步骤
 * 1、声明队列，声明交换机
 * 
 * 2、创建连接
 * 
 * 3、创建通道
 * 
 * 4、通道声明交换机
 * 
 * 5、通道声明队列
 * 
 * 6、通过通道使队列绑定到交换机并指定该队列的routingkey
 * 
 * 7、制定消息
 * 
 * 8、发送消息并指定routingkey
 */
public class RoutingProduceMS {
    //声明两个队列和一个交换机
    //Routing 路由模式
    private static final String QUEUE_EMAIL = "queueEmail";
    private static final String QUEUE_SMS = "queueSms";
    private static final String EXCHANGE = "routingMessageChange";
 
    public static void main(String[] args) {
        Connection connection = null;
        Channel channel = null;
        try {
            ConnectionFactory connectionFactory = new ConnectionFactory();
            connectionFactory.setHost("192.168.43.10");//mq服务ip地址
            connectionFactory.setPort(5672);//mq client连接端口
            connectionFactory.setUsername("guest");//mq登录用户名
            connectionFactory.setPassword("guest");//mq登录密码
            connectionFactory.setVirtualHost("/");//rabbitmq默认虚拟机名称为“/”，虚拟机相当于一个独立的mq服务器
            //创建与RabbitMQ服务的TCP连接
            connection = connectionFactory.newConnection();
            //创建与Exchange的通道，每个连接可以创建多个通道，每个通道代表一个会话任务
            channel = connection.createChannel();
            //通道绑定交换机
            /**
             * 参数明细
             * 1、交换机名称
             * 2、交换机类型，fanout、topic、direct、headers
             */
            //Routing 路由模式
            channel.exchangeDeclare(EXCHANGE, BuiltinExchangeType.DIRECT);
            //通道绑定队列
            /**
             * 声明队列，如果Rabbit中没有此队列将自动创建
             * param1:队列名称
             * param2:是否持久化
             * param3:队列是否独占此连接
             * param4:队列不再使用时是否自动删除此队列
             * param5:队列参数
             * String queue, boolean durable, boolean exclusive, boolean autoDelete, Map arguments
             *
             */
            channel.queueDeclare(QUEUE_EMAIL, true, false, false, null);//通道绑定邮件队列
            channel.queueDeclare(QUEUE_SMS, true, false, false, null);//通道绑定短信队列
            //交换机和队列绑定
            /**
             * 参数明细
             * 1、队列名称
             * 2、交换机名称
             * 3、路由key
             */
            //Routing 路由模式
            channel.queueBind(QUEUE_EMAIL, EXCHANGE, QUEUE_EMAIL);
            channel.queueBind(QUEUE_SMS, EXCHANGE, QUEUE_SMS);
            //给email队列发消息
            for (int i = 0; i < 10; i++) {
                String message = new String("mq 发送email消息。。。");
                /**
                 * 消息发布方法
                 * param1：Exchange的名称，如果没有指定，则使用Default Exchange
                 * param2:routingKey,消息的路由Key，是用于Exchange（交换机）将消息转发到指定的消息队列
                 * param3:消息包含的属性
                 * param4：消息体
                 * 这里没有指定交换机，消息将发送给默认交换机，每个队列也会绑定那个默认的交换机，但是不能显示绑定或解除绑定
                 * 默认的交换机，routingKey等于队列名称
                 */
                //String exchange, String routingKey, BasicProperties props, byte[] body
                //Routing 路由模式
                channel.basicPublish(EXCHANGE, QUEUE_EMAIL, null, message.getBytes());
                System.out.println("mq消息发送成功！");
            }
            //给sms队列发消息
            for (int i = 0; i < 10; i++) {
                String message = new String("mq 发送sms消息。。。");
                /**
                 * 消息发布方法
                 * param1：Exchange的名称，如果没有指定，则使用Default Exchange
                 * param2:routingKey,消息的路由Key，是用于Exchange（交换机）将消息转发到指定的消息队列
                 * param3:消息包含的属性
                 * param4：消息体
                 * 这里没有指定交换机，消息将发送给默认交换机，每个队列也会绑定那个默认的交换机，但是不能显示绑定或解除绑定
                 * 默认的交换机，routingKey等于队列名称
                 */
                //String exchange, String routingKey, BasicProperties props, byte[] body
                //Routing 路由模式
                channel.basicPublish(EXCHANGE, QUEUE_SMS, null, message.getBytes());
                System.out.println("mq消息发送成功！");
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            try {
                channel.close();
            } catch (IOException e) {
                e.printStackTrace();
            } catch (TimeoutException e) {
                e.printStackTrace();
            }
            try {
                connection.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
}

routing路由模式之消费者（邮件）

package com.zzg.rabbitmq.routing;
 
import com.rabbitmq.client.*;
 
import java.io.IOException;
import java.util.concurrent.TimeoutException;
 
/**
 * 消费者核心步骤
 * 1、声明队列，声明交换机
 * 
 * 2、创建连接
 * 
 * 3、创建通道
 * 
 * 4、通道声明交换机
 * 
 * 5、通道声明队列
 * 
 * 6、通过通道使队列绑定到交换机并指定routingkey
 * 
 * 7、重写消息消费方法
 * 
 * 8、执行消息方法
 */
public class RoutingEMailConsumerMS {
    //Routing 路由模式
    private static final String QUEUE_EMAIL = "queueEmail";
    private static final String EXCHANGE = "routingMessageChange";
 
    public static void main(String[] args) {
        Connection connection = null;
        Channel channel = null;
        try {
            ConnectionFactory connectionFactory = new ConnectionFactory();
            connectionFactory.setHost("192.168.43.10");
            connectionFactory.setPort(5672);
            connection = connectionFactory.newConnection();
            channel = connection.createChannel();
            //通道绑定交换机
            /**
             * 参数明细
             * 1、交换机名称
             * 2、交换机类型，fanout、topic、direct、headers
             */
            //Routing 路由模式
            channel.exchangeDeclare(EXCHANGE, BuiltinExchangeType.DIRECT);
            //通道绑定队列
            /**
             * 声明队列，如果Rabbit中没有此队列将自动创建
             * param1:队列名称
             * param2:是否持久化
             * param3:队列是否独占此连接
             * param4:队列不再使用时是否自动删除此队列
             * param5:队列参数
             * String queue, boolean durable, boolean exclusive, boolean autoDelete, Map arguments
             *
             */
            channel.queueDeclare(QUEUE_EMAIL, true, false, false, null);//通道绑定邮件队列
            //交换机和队列绑定
            /**
             * 参数明细
             * 1、队列名称
             * 2、交换机名称
             * 3、路由key
             */
            //Routing 路由模式
            channel.queueBind(QUEUE_EMAIL, EXCHANGE, QUEUE_EMAIL);
            //String consumerTag, Envelope envelope, BasicProperties properties, byte[] body
            DefaultConsumer consumer = new DefaultConsumer(channel) {
                /**
                 * 消费者接收消息调用此方法
                 * @param consumerTag 消费者的标签，在channel.basicConsume()去指定
                 * @param envelope 消息包的内容，可从中获取消息id，消息routingkey，交换机，消息和重传标志
                (收到消息失败后是否需要重新发送)
                 * @param properties
                 * @param body
                 * @throws IOException
                 * String consumerTag, Envelope envelope, BasicProperties properties, byte[] body
                 */
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    //交换机
                    String exchange = envelope.getExchange();
                    //路由key
                    String routingKey = envelope.getRoutingKey();
                    envelope.getDeliveryTag();
                    String msg = new String(body, "utf-8");
                    System.out.println("mq收到的消息是：" + msg);
                }
 
            };
            System.out.println("消费者启动成功！");
            channel.basicConsume(QUEUE_EMAIL, true, consumer);
        } catch (IOException e) {
            e.printStackTrace();
        } catch (TimeoutException e) {
            e.printStackTrace();
        }
    }
}

routing路由模式之消费者（短信）

package com.zzg.rabbitmq.routing;
 
import com.rabbitmq.client.*;
 
import java.io.IOException;
import java.util.concurrent.TimeoutException;
 
/**
 * 消费者核心步骤
 * 1、声明队列，声明交换机
 * 
 * 2、创建连接
 * 
 * 3、创建通道
 * 
 * 4、通道声明交换机
 * 
 * 5、通道声明队列
 * 
 * 6、通过通道使队列绑定到交换机并指定routingkey
 * 
 * 7、重写消息消费方法
 * 
 * 8、执行消息方法
 */
public class RoutingSMSConsumerMS {
    //Routing 路由模式
    private static final String QUEUE_SMS = "queueSms";
    private static final String EXCHANGE = "routingMessageChange";
 
    public static void main(String[] args) {
        Connection connection = null;
        Channel channel = null;
        try {
            ConnectionFactory connectionFactory = new ConnectionFactory();
            connectionFactory.setHost("192.168.43.10");
            connectionFactory.setPort(5672);
            connection = connectionFactory.newConnection();
            channel = connection.createChannel();
            //通道绑定交换机
            /**
             * 参数明细
             * 1、交换机名称
             * 2、交换机类型，fanout、topic、direct、headers
             */
            //Routing 路由模式
            channel.exchangeDeclare(EXCHANGE, BuiltinExchangeType.DIRECT);
            //通道绑定队列
            /**
             * 声明队列，如果Rabbit中没有此队列将自动创建
             * param1:队列名称
             * param2:是否持久化
             * param3:队列是否独占此连接
             * param4:队列不再使用时是否自动删除此队列
             * param5:队列参数
             * String queue, boolean durable, boolean exclusive, boolean autoDelete, Map arguments
             *
             */
            channel.queueDeclare(QUEUE_SMS, true, false, false, null);//通道绑定短信队列
            //交换机和队列绑定
            /**
             * 参数明细
             * 1、队列名称
             * 2、交换机名称
             * 3、路由key
             */
            //Routing 路由模式
            channel.queueBind(QUEUE_SMS, EXCHANGE, QUEUE_SMS);
            DefaultConsumer consumer = new DefaultConsumer(channel) {
                /**
                 * 消费者接收消息调用此方法
                 * @param consumerTag 消费者的标签，在channel.basicConsume()去指定
                 * @param envelope 消息包的内容，可从中获取消息id，消息routingkey，交换机，消息和重传标志
                (收到消息失败后是否需要重新发送)
                 * @param properties
                 * @param body
                 * @throws IOException
                 * String consumerTag, Envelope envelope, BasicProperties properties, byte[] body
                 */
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    //交换机
                    String exchange = envelope.getExchange();
                    //路由key
                    String routingKey = envelope.getRoutingKey();
                    envelope.getDeliveryTag();
                    String msg = new String(body, "utf-8");
                    System.out.println("mq收到的消息是：" + msg);
                }
            };
            System.out.println("消费者启动成功！");
            channel.basicConsume(QUEUE_SMS, true, consumer);
        } catch (IOException e) {
            e.printStackTrace();
        } catch (TimeoutException e) {
            e.printStackTrace();
        }
    }
}

Topics 模式

        Topics 模式和Routing 路由模式最大的区别就是，Topics 模式发送消息和消费消息的时候是通过通配符去进行匹配的。

缺失图片

特点：

1、每个消费者监听自己的队列，并且设置带统配符的routingkey

2、生产者将消息发给broker，由交换机根据routingkey来转发消息到指定的队列

应用场景：用户通知，当用户充值成功或转账完成系统通知用户，通知方式有短信、邮件多种方法；

topics模式之生产者

package com.zzg.rabbitmq.topic;
 
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
 
import java.io.IOException;
import java.util.concurrent.TimeoutException;
 
/**
 * 生产者核心步骤
 * 1、声明队列，声明交换机
 * 
 * 2、创建连接
 * 
 * 3、创建通道
 * 
 * 4、通道声明交换机
 * 
 * 5、通道声明队列
 * 
 * 6、通过通道使队列绑定到交换机并指定该队列的routingkey（通配符）
 * 
 * 7、制定消息
 * 
 * 8、发送消息并指定routingkey（通配符）
 */
public class TopicProduceMS {
    //声明两个队列和一个交换机
    //Topics 模式
    private static final String QUEUE_EMAIL = "queueEmail";
    private static final String QUEUE_SMS = "queueSms";
    private static final String EXCHANGE = "topicMessageChange";
 
    public static void main(String[] args) {
        Connection connection = null;
        Channel channel = null;
        try {
            ConnectionFactory connectionFactory = new ConnectionFactory();
            connectionFactory.setHost("192.168.43.10");//mq服务ip地址
            connectionFactory.setPort(5672);//mq client连接端口
            connectionFactory.setUsername("guest");//mq登录用户名
            connectionFactory.setPassword("guest");//mq登录密码
            connectionFactory.setVirtualHost("/");//rabbitmq默认虚拟机名称为“/”，虚拟机相当于一个独立的mq服务器
            //创建与RabbitMQ服务的TCP连接
            connection = connectionFactory.newConnection();
            //创建与Exchange的通道，每个连接可以创建多个通道，每个通道代表一个会话任务
            channel = connection.createChannel();
            //通道绑定交换机
            /**
             * 参数明细
             * 1、交换机名称
             * 2、交换机类型，fanout、topic、direct、headers
             */
            //Topics 模式
            channel.exchangeDeclare(EXCHANGE, BuiltinExchangeType.TOPIC);
            //通道绑定队列
            /**
             * 声明队列，如果Rabbit中没有此队列将自动创建
             * param1:队列名称
             * param2:是否持久化
             * param3:队列是否独占此连接
             * param4:队列不再使用时是否自动删除此队列
             * param5:队列参数
             * String queue, boolean durable, boolean exclusive, boolean autoDelete, Map arguments
             *
             */
            channel.queueDeclare(QUEUE_EMAIL, true, false, false, null);//通道绑定邮件队列
            channel.queueDeclare(QUEUE_SMS, true, false, false, null);//通道绑定短信队列
            //交换机和队列绑定
            /**
             * 参数明细
             * 1、队列名称
             * 2、交换机名称
             * 3、路由key
             */
            channel.queueBind(QUEUE_EMAIL, EXCHANGE, "inform.#.email.#");
            channel.queueBind(QUEUE_SMS, EXCHANGE, "inform.#.sms.#");
            //给email队列发消息
            for (int i = 0; i < 10; i++) {
                String message = new String("mq 发送email消息。。。");
                /**
                 * 消息发布方法
                 * param1：Exchange的名称，如果没有指定，则使用Default Exchange
                 * param2:routingKey,消息的路由Key，是用于Exchange（交换机）将消息转发到指定的消息队列
                 * param3:消息包含的属性
                 * param4：消息体
                 * 这里没有指定交换机，消息将发送给默认交换机，每个队列也会绑定那个默认的交换机，但是不能显示绑定或解除绑定
                 * 默认的交换机，routingKey等于队列名称
                 */
                //String exchange, String routingKey, BasicProperties props, byte[] body
                channel.basicPublish(EXCHANGE, "inform.email", null, message.getBytes());
                System.out.println("mq email 消息发送成功！");
            }
            //给sms队列发消息
            for (int i = 0; i < 10; i++) {
                String message = new String("mq 发送sms消息。。。");
                /**
                 * 消息发布方法
                 * param1：Exchange的名称，如果没有指定，则使用Default Exchange
                 * param2:routingKey,消息的路由Key，是用于Exchange（交换机）将消息转发到指定的消息队列
                 * param3:消息包含的属性
                 * param4：消息体
                 * 这里没有指定交换机，消息将发送给默认交换机，每个队列也会绑定那个默认的交换机，但是不能显示绑定或解除绑定
                 * 默认的交换机，routingKey等于队列名称
                 */
                //String exchange, String routingKey, BasicProperties props, byte[] body
                channel.basicPublish(EXCHANGE, "inform.sms", null, message.getBytes());
                System.out.println("mq sms 消息发送成功！");
            }
            //给email和sms队列发消息
            for (int i = 0; i < 10; i++) {
                String message = new String("mq 发送email sms消息。。。");
                /**
                 * 消息发布方法
                 * param1：Exchange的名称，如果没有指定，则使用Default Exchange
                 * param2:routingKey,消息的路由Key，是用于Exchange（交换机）将消息转发到指定的消息队列
                 * param3:消息包含的属性
                 * param4：消息体
                 * 这里没有指定交换机，消息将发送给默认交换机，每个队列也会绑定那个默认的交换机，但是不能显示绑定或解除绑定
                 * 默认的交换机，routingKey等于队列名称
                 */
                //String exchange, String routingKey, BasicProperties props, byte[] body
                channel.basicPublish(EXCHANGE, "inform.email.sms", null, message.getBytes());
                System.out.println("mq email sms 消息发送成功！");
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            try {
                channel.close();
            } catch (IOException e) {
                e.printStackTrace();
            } catch (TimeoutException e) {
                e.printStackTrace();
            }
            try {
                connection.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
 
    }
}

topics模式之消费者（邮件）

package com.zzg.rabbitmq.topic;
 
import com.rabbitmq.client.*;
 
import java.io.IOException;
import java.util.concurrent.TimeoutException;
 
/**
 * 消费者核心步骤
 */
public class TopicEMailConsumerMS {
    private static final String QUEUE_EMAIL = "queueEmail";
    private static final String EXCHANGE = "topicMessageChange";
 
    public static void main(String[] args) {
        Connection connection = null;
        Channel channel = null;
        try {
            ConnectionFactory connectionFactory = new ConnectionFactory();
            connectionFactory.setHost("192.168.43.10");
            connectionFactory.setPort(5672);
            connection = connectionFactory.newConnection();
            channel = connection.createChannel();
            //通道绑定交换机
            /**
             * 参数明细
             * 1、交换机名称
             * 2、交换机类型，fanout、topic、direct、headers
             */
            channel.exchangeDeclare(EXCHANGE, BuiltinExchangeType.TOPIC);
            //通道绑定队列
            /**
             * 声明队列，如果Rabbit中没有此队列将自动创建
             * param1:队列名称
             * param2:是否持久化
             * param3:队列是否独占此连接
             * param4:队列不再使用时是否自动删除此队列
             * param5:队列参数
             * String queue, boolean durable, boolean exclusive, boolean autoDelete, Map arguments
             *
             */
            channel.queueDeclare(QUEUE_EMAIL, true, false, false, null);//通道绑定邮件队列
            //交换机和队列绑定
            /**
             * 参数明细
             * 1、队列名称
             * 2、交换机名称
             * 3、路由key
             */
            channel.queueBind(QUEUE_EMAIL, EXCHANGE, "inform.#.email.#");
            //String consumerTag, Envelope envelope, BasicProperties properties, byte[] body
            DefaultConsumer consumer = new DefaultConsumer(channel) {
                /**
                 * 消费者接收消息调用此方法
                 * @param consumerTag 消费者的标签，在channel.basicConsume()去指定
                 * @param envelope 消息包的内容，可从中获取消息id，消息routingkey，交换机，消息和重传标志
                (收到消息失败后是否需要重新发送)
                 * @param properties
                 * @param body
                 * @throws IOException
                 * String consumerTag, Envelope envelope, BasicProperties properties, byte[] body
                 */
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    //交换机
                    String exchange = envelope.getExchange();
                    //路由key
                    String routingKey = envelope.getRoutingKey();
                    envelope.getDeliveryTag();
                    String msg = new String(body, "utf-8");
                    System.out.println("mq收到的消息是：" + msg);
                }
 
            };
            System.out.println("消费者启动成功！");
            channel.basicConsume(QUEUE_EMAIL, true, consumer);
        } catch (IOException e) {
            e.printStackTrace();
        } catch (TimeoutException e) {
            e.printStackTrace();
        }
    }
}