DirectProducer： direct 类型交换器的生产者
NormalConsumer： 普通的消费者
MultiBindConsumer： 队列绑定到交换器上时，是允许绑定多个路由键的，也就是多重绑定
对比：单个绑定的消费者只能收到指定的消息，多重绑定的的消费者可以收到所有的消息

生产者

  3. import com.rabbitmq.client.BuiltinExchangeType;
  4. import com.rabbitmq.client.Channel;
  5. import com.rabbitmq.client.Connection;
  6. import com.rabbitmq.client.ConnectionFactory;
  8. import java.io.IOException;
  9. import java.util.concurrent.TimeoutException;
  11. /**
  12.  * 类说明：direct类型交换器的生产者
  13.  */
  14. public class DirectProducer {
  15.     //声明交换器名字
  16.     public final static String EXCHANGE_NAME = "direct_logs";
  18.     public static void main(String[] args)
  19.             throws IOException, TimeoutException {
  20.         //创建连接、连接到RabbitMQ
  21.         ConnectionFactory connectionFactory = new ConnectionFactory();
  22.         //设置下连接工厂的连接地址(使用默认端口5672)
  23.         connectionFactory.setHost("zjj101");
  24.         /*设置登陆用户名和密码*/
  25.         connectionFactory.setUsername("guest");
  26.         connectionFactory.setPassword("guest");
  27.         /*设置虚拟主机地址,不能设置错了,不然会报错误*/
  28.         connectionFactory.setVirtualHost("/");
  29. //        connectionFactory.setHost("localhost");
  30.         //创建连接
  31.         Connection connection = connectionFactory.newConnection();
  32.         //创建信道
  33.         Channel channel = connection.createChannel();
  35.         //在信道中设置交换器 ,交换器是枚举
  36.         channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT);
  38.         //申明队列（放在消费者中去做）
  39.         //申明路由键\消息体
  40.         String[] routeKeys = {"king", "mark", "james"};
  41.         for (int i = 0; i < 6; i++) {
  42.             String routeKey = routeKeys[i % 3];
  43.             String msg = "Hello,RabbitMQ" + (i + 1);
  44.             //发布消息
  45.             channel.basicPublish(EXCHANGE_NAME, routeKey, null, msg.getBytes());
  46.             System.out.println("Sent:" + routeKey + ":" + msg);
  47.         }
  48.         channel.close();
  49.         connection.close();
  50.     }
  52. }

消费者1

  2. import com.rabbitmq.client.*;
  3. import java.io.IOException;
  4. import java.util.concurrent.TimeoutException;
  6. /**
  7.  * 类说明：队列和交换器的多重绑定
  8.  * 多个路由键的绑定
  9.  * 案例:ZJJ_RabbitMQ_2019/11/03_13:17:45_y9y2f
  10.  */
  11. public class MultiBindConsumer {
  13.     public static void main(String[] argv) throws IOException, TimeoutException {
  14.         //连接工厂
  15.         ConnectionFactory factory = new ConnectionFactory();
  16.         //连接rabbitMq的地址
  17.         factory.setHost("zjj101");
  18.         /*设置登陆用户名和密码*/
  19.         factory.setUsername("guest");
  20.         factory.setPassword("guest");
  21.         /*设置虚拟主机地址,不能设置错了,不然会报错误*/
  22.         factory.setVirtualHost("/");
  23.         // 打开连接和创建频道，与发送端一样
  24.         Connection connection = factory.newConnection();
  25.         //创建一个信道
  26.         final Channel channel = connection.createChannel();
  27.         //信道设置交换器类型(direct)
  28.         channel.exchangeDeclare(DirectProducer.EXCHANGE_NAME, BuiltinExchangeType.DIRECT);
  30.         //声明一个随机队列,每一次生成的队列都是不同的.
  31.         String queueName = channel.queueDeclare().getQueue();
  32.         /*队列绑定到交换器上时，是允许绑定多个路由键的，也就是多重绑定
  33.          * 做法就是for循环执行多次方法
  34.          * */
  35.         String[] routekeys = {"king", "mark", "james"};
  36.         for (String routekey : routekeys) {
  37.             channel.queueBind(queueName, DirectProducer.EXCHANGE_NAME,
  38.                     routekey);
  39.         }
  40.         System.out.println(" [*] 等待消息:");
  42.         // 创建队列消费者
  43.         final Consumer consumerA = new DefaultConsumer(channel) {
  44.             @Override
  45.             public void handleDelivery(String consumerTag,
  46.                                        Envelope envelope,
  47.                                        AMQP.BasicProperties
  48.                                                properties,
  49.                                        byte[] body)
  50.                     throws IOException {
  51.                 String message = new String(body, "UTF-8");
  52.                 System.out.println(" 接收到routingKey为  "
  53.                         + envelope.getRoutingKey() + ":'" + message
  54.                         + "'");
  55.             }
  56.         };
  57.         channel.basicConsume(queueName, true, consumerA);
  58.     }
  59. }

输出:

  1.  [*] 等待消息:
  2.  接收到routingKey为  king:'Hello,RabbitMQ1'
  3.  接收到routingKey为  mark:'Hello,RabbitMQ2'
  4.  接收到routingKey为  james:'Hello,RabbitMQ3'
  5.  接收到routingKey为  king:'Hello,RabbitMQ4'
  6.  接收到routingKey为  mark:'Hello,RabbitMQ5'
  7.  接收到routingKey为  james:'Hello,RabbitMQ6'

消费者2

  2. import com.rabbitmq.client.*;
  3. import java.io.IOException;
  4. import java.util.concurrent.TimeoutException;
  6. /**
  7.  * 类说明：普通的消费者
  8.  */
  9. public class NormalConsumer {
  11.     public static void main(String[] argv)
  12.             throws IOException, TimeoutException {
  13.         //创建连接、连接到RabbitMQ
  14.         ConnectionFactory connectionFactory = new ConnectionFactory();
  15.         //设置下连接工厂的连接地址(使用默认端口5672)
  16.         connectionFactory.setHost("zjj101");
  17.         /*设置登陆用户名和密码*/
  18.         connectionFactory.setUsername("guest");
  19.         connectionFactory.setPassword("guest");
  20.         /*设置虚拟主机地址,不能设置错了,不然会报错误*/
  21.         connectionFactory.setVirtualHost("/");
  22.         //创建连接
  23.         Connection connection = connectionFactory.newConnection();
  24.         //创建信道
  26.         Channel channel = connection.createChannel();
  28.         //在信道中设置交换器
  29.         channel.exchangeDeclare(DirectProducer.EXCHANGE_NAME, BuiltinExchangeType.DIRECT);
  31.         //申明队列（放在消费者中去做）
  32.         String queueName = "queue-king";
  33.         channel.queueDeclare(queueName, false, false, false, null);
  35.         //绑定：将队列(queuq-king)与交换器通过 路由键 绑定(king)
  36.         String routeKey = "king";
  37.         channel.queueBind(queueName, DirectProducer.EXCHANGE_NAME, routeKey);
  38.         System.out.println("等待接收消息 ......");
  40.         //申明一个消费者
  41.         final Consumer consumer = new DefaultConsumer(channel) {
  42.             /**
  43.              *
  44.              * @param s
  45.              * @param envelope
  46.              * @param basicProperties
  47.              * @param bytes 是生产者发送过来的消息
  48.              * @throws IOException
  49.              */
  50.             @Override
  51.             public void handleDelivery(String s, Envelope envelope, AMQP.BasicProperties basicProperties, byte[] bytes) throws IOException {
  52.                 String message = new String(bytes, "UTF-8");
  53.                 System.out.println("收到routingKey为:[" + envelope.getRoutingKey() + "]的消息,消息内容是: " + message);
  54.             }
  55.         };
  56.         //消息者正是开始在指定队列上消费。(queue-king)
  57.         /*
  58.          * 参数1: 队列名称
  59.          * 参数2:自动确认参数，如果是true则是自动确认,意思是我每一次消费了,
  60.          *         我需要给RabbitMQ确认一下我已经消费完了
  61.          * 参数3:消费者
  62.          */
  63.         channel.basicConsume(queueName, true, consumer);
  66.     }
  68. }

输出:

  1. 等待接收消息 ......
  2. 收到routingKey为:[king]的消息,消息内容是: Hello,RabbitMQ1
  3. 收到routingKey为:[king]的消息,消息内容是: Hello,RabbitMQ4