DirectProducer： direct 类型交换器的生产者
MultiChannelConsumer： 一个连接下允许有多个信道
一个连接，我们可以使用多线程的方式模拟多个信道进行通讯。这样可以做到多路复用。

生产者

  2. import com.rabbitmq.client.BuiltinExchangeType;
  3. import com.rabbitmq.client.Channel;
  4. import com.rabbitmq.client.Connection;
  5. import com.rabbitmq.client.ConnectionFactory;
  7. import java.io.IOException;
  8. import java.util.concurrent.TimeoutException;
  10. /**
  11.  * 类说明：direct类型交换器的生产者
  12.  * 案例:ZJJ_RabbitMQ_2019/11/03_13:17:17_fk6sz
  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. Sent:king:Hello,RabbitMQ1
  2. Sent:mark:Hello,RabbitMQ2
  3. Sent:james:Hello,RabbitMQ3
  4. Sent:king:Hello,RabbitMQ4
  5. Sent:mark:Hello,RabbitMQ5
  6. Sent:james:Hello,RabbitMQ6

消费者

  2. import com.rabbitmq.client.*;
  3. import java.io.IOException;
  4. import java.util.concurrent.TimeoutException;
  6. /**
  7.  * 类说明：一个连接多个信道
  8.  * 使用了多线程的支持
  9.  * 演示消费者使用一个连接两个信道
  10.  * 案例:ZJJ_RabbitMQ_2019/11/03_13:19:03_2vcj5
  11.  */
  12. public class MultiChannelConsumer {
  14.     private static class ConsumerWorker implements Runnable {
  15.         final Connection connection;
  17.         public ConsumerWorker(Connection connection) {
  18.             this.connection = connection;
  19.         }
  21.         public void run() {
  22.             try {
  23.                 /*创建一个信道，意味着每个线程单独一个信道*/
  24.                 Channel channel = connection.createChannel();
  25.                 //信道设置交换器类型(direct)
  26.                 channel.exchangeDeclare(DirectProducer.EXCHANGE_NAME, BuiltinExchangeType.DIRECT);
  27.                 // 声明一个随机队列
  28.                 String queueName = channel.queueDeclare().getQueue();
  29.                 //String queueName = "queue-king";      // 同一个队列
  31.                 //消费者名字，打印输出用
  32.                 final String consumerName = Thread.currentThread().getName() + "-all";
  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("[" + consumerName + "] Waiting for messages:");
  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 =
  52.                                 new String(body, "UTF-8");
  53.                         System.out.println(consumerName
  54.                                 + " Received " + envelope.getRoutingKey()
  55.                                 + ":'" + message + "'");
  56.                     }
  57.                 };
  58.                 channel.basicConsume(queueName, true, consumerA);
  59.             } catch (Exception e) {
  60.                 e.printStackTrace();
  61.             }
  62.         }
  63.     }
  65.     public static void main(String[] argv) throws IOException,
  66.             InterruptedException, TimeoutException {
  67.         //连接工厂
  68.         ConnectionFactory factory = new ConnectionFactory();
  69.         //连接rabbitMq的地址
  71.         factory.setHost("zjj101");
  72.         /*设置登陆用户名和密码*/
  73.         factory.setUsername("guest");
  74.         factory.setPassword("guest");
  75.         /*设置虚拟主机地址,不能设置错了,不然会报错误*/
  76.         factory.setVirtualHost("/");
  78.         // 打开连接和创建频道，与发送端一样
  79.         Connection connection = factory.newConnection();
  80.         //一个连接多个信道
  81.         for (int i = 0; i < 2; i++) {
  82.             /*将连接作为参数，传递给每个线程*/
  83.             Thread worker = new Thread(new ConsumerWorker(connection));
  84.             worker.start();
  85.         }
  86.     }
  87. }

输出:

  1. [Thread-0-all] Waiting for messages:
  2. [Thread-1-all] Waiting for messages:
  3. Thread-1-all Received king:'Hello,RabbitMQ1'
  4. Thread-1-all Received mark:'Hello,RabbitMQ2'
  5. Thread-1-all Received james:'Hello,RabbitMQ3'
  6. Thread-0-all Received king:'Hello,RabbitMQ1'
  7. Thread-0-all Received mark:'Hello,RabbitMQ2'
  8. Thread-0-all Received james:'Hello,RabbitMQ3'
  9. Thread-0-all Received king:'Hello,RabbitMQ4'
  10. Thread-0-all Received mark:'Hello,RabbitMQ5'
  11. Thread-0-all Received james:'Hello,RabbitMQ6'
  12. Thread-1-all Received king:'Hello,RabbitMQ4'
  13. Thread-1-all Received mark:'Hello,RabbitMQ5'
  14. Thread-1-all Received james:'Hello,RabbitMQ6'