生产者

  1. import com.rabbitmq.client.BuiltinExchangeType;
  2. import com.rabbitmq.client.Channel;
  3. import com.rabbitmq.client.Connection;
  4. import com.rabbitmq.client.ConnectionFactory;
  6. import java.io.IOException;
  7. import java.util.concurrent.TimeoutException;
  9. /**
  10.  * 类说明：Topic类型的生产者
  11.  * 假设有交换器 topic_course，
  12.  * 讲课老师有king,mark,james，
  13.  * 技术专题有kafka,jvm,redis，
  14.  * 课程章节有 A、B、C，
  15.  * 路由键的规则为 讲课老师+“.”+技术专题+“.”+课程章节，如：king.kafka.A。
  16.  * 生产者--生产全部的消息3*3*3=27条消息
  17.  */
  18. public class TopicProducer {
  20.     public final static String EXCHANGE_NAME = "topic_course";
  22.     public static void main(String[] args)
  23.             throws IOException, TimeoutException {
  24.         /**
  25.          * 创建连接连接到RabbitMQ
  26.          */
  27.         ConnectionFactory factory = new ConnectionFactory();
  28.         //连接rabbitMq的地址
  29.         factory.setHost("zjj101");
  30.         /*设置登陆用户名和密码*/
  31.         factory.setUsername("guest");
  32.         factory.setPassword("guest");
  33.         /*设置虚拟主机地址,不能设置错了,不然会报错误*/
  34.         factory.setVirtualHost("/");
  35.         // 创建一个连接
  36.         Connection connection = factory.newConnection();
  38.         // 创建一个信道
  39.         Channel channel = connection.createChannel();
  40.         channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.TOPIC);
  42.         /*我们的课程，路由键最终格式类似于：king.kafka.A  king.kafka.B*/
  43.         String[] techers = {"king", "mark", "james"};
  44.         for (int i = 0; i < 3; i++) {
  45.             String[] modules = {"kafka", "jvm", "redis"};
  46.             for (int j = 0; j < 3; j++) {
  47.                 String[] servers = {"A", "B", "C"};
  48.                 for (int k = 0; k < 3; k++) {
  49.                     // 发送的消息
  50.                     String message = "Hello Topic_[" + i + "," + j + "," + k + "]";
  51.                     String routeKey = techers[i % 3] + "." + modules[j % 3]
  52.                             + "." + servers[k % 3];
  53.                     channel.basicPublish(EXCHANGE_NAME, routeKey,
  54.                             null, message.getBytes());
  55.                     System.out.println(" [x] Sent '" + routeKey + ":'"
  56.                             + message + "'");
  57.                 }
  58.             }
  60.         }
  61.         // 关闭频道和连接
  62.         channel.close();
  63.         connection.close();
  64.     }
  66. }

输出:

  1.  [x] Sent 'king.kafka.A:'Hello Topic_[0,0,0]'
  2.  [x] Sent 'king.kafka.B:'Hello Topic_[0,0,1]'
  3.  [x] Sent 'king.kafka.C:'Hello Topic_[0,0,2]'
  4.  [x] Sent 'king.jvm.A:'Hello Topic_[0,1,0]'
  5.  [x] Sent 'king.jvm.B:'Hello Topic_[0,1,1]'
  6.  [x] Sent 'king.jvm.C:'Hello Topic_[0,1,2]'
  7.  [x] Sent 'king.redis.A:'Hello Topic_[0,2,0]'
  8.  [x] Sent 'king.redis.B:'Hello Topic_[0,2,1]'
  9.  [x] Sent 'king.redis.C:'Hello Topic_[0,2,2]'
  10.  [x] Sent 'mark.kafka.A:'Hello Topic_[1,0,0]'
  11.  [x] Sent 'mark.kafka.B:'Hello Topic_[1,0,1]'
  12.  [x] Sent 'mark.kafka.C:'Hello Topic_[1,0,2]'
  13.  [x] Sent 'mark.jvm.A:'Hello Topic_[1,1,0]'
  14.  [x] Sent 'mark.jvm.B:'Hello Topic_[1,1,1]'
  15.  [x] Sent 'mark.jvm.C:'Hello Topic_[1,1,2]'
  16.  [x] Sent 'mark.redis.A:'Hello Topic_[1,2,0]'
  17.  [x] Sent 'mark.redis.B:'Hello Topic_[1,2,1]'
  18.  [x] Sent 'mark.redis.C:'Hello Topic_[1,2,2]'
  19.  [x] Sent 'james.kafka.A:'Hello Topic_[2,0,0]'
  20.  [x] Sent 'james.kafka.B:'Hello Topic_[2,0,1]'
  21.  [x] Sent 'james.kafka.C:'Hello Topic_[2,0,2]'
  22.  [x] Sent 'james.jvm.A:'Hello Topic_[2,1,0]'
  23.  [x] Sent 'james.jvm.B:'Hello Topic_[2,1,1]'
  24.  [x] Sent 'james.jvm.C:'Hello Topic_[2,1,2]'
  25.  [x] Sent 'james.redis.A:'Hello Topic_[2,2,0]'
  26.  [x] Sent 'james.redis.B:'Hello Topic_[2,2,1]'
  27.  [x] Sent 'james.redis.C:'Hello Topic_[2,2,2]'

消费者- 消费所有的消息

  2. import com.rabbitmq.client.*;
  3. import java.io.IOException;
  4. import java.util.concurrent.TimeoutException;
  6. /**
  7.  * 1、要关注所有的课程，怎么做？ #
  8.  * #  可以匹配多个
  9.  * 类说明：所有的课程内容
  10.  */
  11. public class AllConsumer {
  13.     public static void main(String[] argv) throws IOException,
  14.             InterruptedException, TimeoutException {
  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.         final Channel channel = connection.createChannel();
  27.         channel.exchangeDeclare(TopicProducer.EXCHANGE_NAME,
  28.                 BuiltinExchangeType.TOPIC);
  29.         // 声明一个随机队列
  30.         String queueName = channel.queueDeclare().getQueue();
  32.         //!!!!!!!!!消费了所有的消息  #
  33.         channel.queueBind(queueName, TopicProducer.EXCHANGE_NAME, "#");
  35.         System.out.println(" [*] Waiting for messages:");
  37.         // 创建队列消费者
  38.         final Consumer consumerA = new DefaultConsumer(channel) {
  39.             @Override
  40.             public void handleDelivery(String consumerTag,
  41.                                        Envelope envelope,
  42.                                        AMQP.BasicProperties properties,
  43.                                        byte[] body)
  44.                     throws IOException {
  45.                 String message = new String(body, "UTF-8");
  46.                 System.out.println(" AllConsumer Received "
  47.                         + envelope.getRoutingKey()
  48.                         + "':'" + message + "'");
  49.             }
  50.         };
  51.         channel.basicConsume(queueName, true, consumerA);
  52.     }
  53. }

启动完了消费者后,启动生产者, 输出日志结果:

所有的消息都被消费了.

  1.  [*] Waiting for messages:
  2.  AllConsumer Received king.kafka.A':'Hello Topic_[0,0,0]'
  3.  AllConsumer Received king.kafka.B':'Hello Topic_[0,0,1]'
  4.  AllConsumer Received king.kafka.C':'Hello Topic_[0,0,2]'
  5.  AllConsumer Received king.jvm.A':'Hello Topic_[0,1,0]'
  6.  AllConsumer Received king.jvm.B':'Hello Topic_[0,1,1]'
  7.  AllConsumer Received king.jvm.C':'Hello Topic_[0,1,2]'
  8.  AllConsumer Received king.redis.A':'Hello Topic_[0,2,0]'
  9.  AllConsumer Received king.redis.B':'Hello Topic_[0,2,1]'
  10.  AllConsumer Received king.redis.C':'Hello Topic_[0,2,2]'
  11.  AllConsumer Received mark.kafka.A':'Hello Topic_[1,0,0]'
  12.  AllConsumer Received mark.kafka.B':'Hello Topic_[1,0,1]'
  13.  AllConsumer Received mark.kafka.C':'Hello Topic_[1,0,2]'
  14.  AllConsumer Received mark.jvm.A':'Hello Topic_[1,1,0]'
  15.  AllConsumer Received mark.jvm.B':'Hello Topic_[1,1,1]'
  16.  AllConsumer Received mark.jvm.C':'Hello Topic_[1,1,2]'
  17.  AllConsumer Received mark.redis.A':'Hello Topic_[1,2,0]'
  18.  AllConsumer Received mark.redis.B':'Hello Topic_[1,2,1]'
  19.  AllConsumer Received mark.redis.C':'Hello Topic_[1,2,2]'
  20.  AllConsumer Received james.kafka.A':'Hello Topic_[2,0,0]'
  21.  AllConsumer Received james.kafka.B':'Hello Topic_[2,0,1]'
  22.  AllConsumer Received james.kafka.C':'Hello Topic_[2,0,2]'
  23.  AllConsumer Received james.jvm.A':'Hello Topic_[2,1,0]'
  24.  AllConsumer Received james.jvm.B':'Hello Topic_[2,1,1]'
  25.  AllConsumer Received james.jvm.C':'Hello Topic_[2,1,2]'
  26.  AllConsumer Received james.redis.A':'Hello Topic_[2,2,0]'
  27.  AllConsumer Received james.redis.B':'Hello Topic_[2,2,1]'
  28.  AllConsumer Received james.redis.C':'Hello Topic_[2,2,2]'

消费者-指定前缀打头

  2. import com.rabbitmq.client.*;
  4. import java.io.IOException;
  5. import java.util.concurrent.TimeoutException;
  7. /**
  8.  2、关注king老师的所有课程，怎么办？
  9.  *类说明：
  10.  */
  11. public class King_AllConsumer {
  13.     public static void main(String[] argv) throws IOException,
  14.             InterruptedException, TimeoutException {
  15.         ConnectionFactory factory = new ConnectionFactory();
  16.         //连接rabbitMq的地址
  17.         factory.setHost("zjj101");
  18.         /*设置登陆用户名和密码*/
  19.         factory.setUsername("guest");
  20.         factory.setPassword("guest");
  21.         /*设置虚拟主机地址,不能设置错了,不然会报错误*/
  22.         factory.setVirtualHost("/");
  24.         // 打开连接和创建频道，与发送端一样
  25.         Connection connection = factory.newConnection();
  26.         final Channel channel = connection.createChannel();
  28.         channel.exchangeDeclare(TopicProducer.EXCHANGE_NAME,
  29.                 BuiltinExchangeType.TOPIC);
  30.         // 声明一个随机队列
  31.         String queueName = channel.queueDeclare().getQueue();
  32.         //TODO
  33.         channel.queueBind(queueName,TopicProducer.EXCHANGE_NAME, "king.#");
  35.         System.out.println(" [*] Waiting for messages:");
  37.         // 创建队列消费者
  38.         final Consumer consumerA = new DefaultConsumer(channel) {
  39.             @Override
  40.             public void handleDelivery(String consumerTag,
  41.                                        Envelope envelope,
  42.                                        AMQP.BasicProperties properties,
  43.                                        byte[] body)
  44.                     throws IOException {
  45.                 String message = new String(body, "UTF-8");
  46.                 System.out.println(" AllConsumer Received "
  47.                         + envelope.getRoutingKey()
  48.                         + "':'" + message + "'");
  49.             }
  50.         };
  51.         channel.basicConsume(queueName, true, consumerA);
  52.     }
  53. }

启动完这一个消费者再启动生产者后输出结果:

king打头的routingKey都被消费了

  1.  [*] Waiting for messages:
  2.  AllConsumer Received king.kafka.A':'Hello Topic_[0,0,0]'
  3.  AllConsumer Received king.kafka.B':'Hello Topic_[0,0,1]'
  4.  AllConsumer Received king.kafka.C':'Hello Topic_[0,0,2]'
  5.  AllConsumer Received king.jvm.A':'Hello Topic_[0,1,0]'
  6.  AllConsumer Received king.jvm.B':'Hello Topic_[0,1,1]'
  7.  AllConsumer Received king.jvm.C':'Hello Topic_[0,1,2]'
  8.  AllConsumer Received king.redis.A':'Hello Topic_[0,2,0]'
  9.  AllConsumer Received king.redis.B':'Hello Topic_[0,2,1]'
  10.  AllConsumer Received king.redis.C':'Hello Topic_[0,2,2]'

消费者-指定前缀和结尾

  1. import com.rabbitmq.client.*;
  3. import java.io.IOException;
  4. import java.util.concurrent.TimeoutException;
  6. /**
  7.  3、关注king老师所有的A章节，怎么办？
  8.  *类说明：
  9.  */
  10. public class King_All_AConsumer {
  12.     public static void main(String[] argv) throws IOException,
  13.             InterruptedException, TimeoutException {
  14.         ConnectionFactory factory = new ConnectionFactory();
  15.         //连接rabbitMq的地址
  16.         factory.setHost("zjj101");
  17.         /*设置登陆用户名和密码*/
  18.         factory.setUsername("guest");
  19.         factory.setPassword("guest");
  20.         /*设置虚拟主机地址,不能设置错了,不然会报错误*/
  21.         factory.setVirtualHost("/");
  23.         // 打开连接和创建频道，与发送端一样
  24.         Connection connection = factory.newConnection();
  25.         final Channel channel = connection.createChannel();
  27.         channel.exchangeDeclare(TopicProducer.EXCHANGE_NAME,
  28.                 BuiltinExchangeType.TOPIC);
  29.         // 声明一个随机队列
  30.         String queueName = channel.queueDeclare().getQueue();
  31.         //!!!!!!!!!!!!配置匹配
  32.         channel.queueBind(queueName,TopicProducer.EXCHANGE_NAME, "king.*.A");
  34.         System.out.println(" [*] Waiting for messages:");
  36.         // 创建队列消费者
  37.         final Consumer consumerA = new DefaultConsumer(channel) {
  38.             @Override
  39.             public void handleDelivery(String consumerTag,
  40.                                        Envelope envelope,
  41.                                        AMQP.BasicProperties properties,
  42.                                        byte[] body)
  43.                     throws IOException {
  44.                 String message = new String(body, "UTF-8");
  45.                 System.out.println(" AllConsumer Received "
  46.                         + envelope.getRoutingKey()
  47.                         + "':'" + message + "'");
  48.             }
  49.         };
  50.         channel.basicConsume(queueName, true, consumerA);
  51.     }
  52. }

启动完这一个消费者再启动生产者后输出结果:

king打头的并且 A结尾的 routingKey都被消费了

  1.  [*] Waiting for messages:
  2.  AllConsumer Received king.kafka.A':'Hello Topic_[0,0,0]'
  3.  AllConsumer Received king.jvm.A':'Hello Topic_[0,1,0]'
  4.  AllConsumer Received king.redis.A':'Hello Topic_[0,2,0]'

消费者-指定中间内容

  2. import com.rabbitmq.client.*;
  3. import java.io.IOException;
  4. import java.util.concurrent.TimeoutException;
  6. /**
  7.  4、关注kafka所有的课程，怎么办？
  8.  *类说明：
  9.  */
  10. public class Kafka_AllConsumer {
  12.     public static void main(String[] argv) throws IOException,
  13.             InterruptedException, TimeoutException {
  14.         ConnectionFactory factory = new ConnectionFactory();
  15.         //连接rabbitMq的地址
  16.         factory.setHost("zjj101");
  17.         /*设置登陆用户名和密码*/
  18.         factory.setUsername("guest");
  19.         factory.setPassword("guest");
  20.         /*设置虚拟主机地址,不能设置错了,不然会报错误*/
  21.         factory.setVirtualHost("/");
  23.         // 打开连接和创建频道，与发送端一样
  24.         Connection connection = factory.newConnection();
  25.         final Channel channel = connection.createChannel();
  27.         channel.exchangeDeclare(TopicProducer.EXCHANGE_NAME,
  28.                 BuiltinExchangeType.TOPIC);
  29.         // 声明一个随机队列
  30.         String queueName = channel.queueDeclare().getQueue();
  32.         channel.queueBind(queueName,TopicProducer.EXCHANGE_NAME,
  33.                 "#.kafka.#");
  35.         System.out.println(" [*] Waiting for messages:");
  37.         // 创建队列消费者
  38.         final Consumer consumerA = new DefaultConsumer(channel) {
  39.             @Override
  40.             public void handleDelivery(String consumerTag,
  41.                                        Envelope envelope,
  42.                                        AMQP.BasicProperties properties,
  43.                                        byte[] body)
  44.                     throws IOException {
  45.                 String message = new String(body, "UTF-8");
  46.                 System.out.println(" AllConsumer Received "
  47.                         + envelope.getRoutingKey()
  48.                         + "':'" + message + "'");
  49.             }
  50.         };
  51.         channel.basicConsume(queueName, true, consumerA);
  52.     }
  53. }

启动完这一个消费者再启动生产者后输出结果:

中间为Kafka的的 routingKey都被消费了

  1. [*] Waiting for messages:
  2.  AllConsumer Received king.kafka.A':'Hello Topic_[0,0,0]'
  3.  AllConsumer Received king.kafka.B':'Hello Topic_[0,0,1]'
  4.  AllConsumer Received king.kafka.C':'Hello Topic_[0,0,2]'
  5.  AllConsumer Received mark.kafka.A':'Hello Topic_[1,0,0]'
  6.  AllConsumer Received mark.kafka.B':'Hello Topic_[1,0,1]'
  7.  AllConsumer Received mark.kafka.C':'Hello Topic_[1,0,2]'
  8.  AllConsumer Received james.kafka.A':'Hello Topic_[2,0,0]'
  9.  AllConsumer Received james.kafka.B':'Hello Topic_[2,0,1]'
  10.  AllConsumer Received james.kafka.C':'Hello Topic_[2,0,2]'

消费者-指定结尾

  1. import com.rabbitmq.client.*;
  3. import java.io.IOException;
  4. import java.util.concurrent.TimeoutException;
  6. /**
  7.  * 5、关注所有的B章节，怎么办？
  8.  * 类说明：
  9.  */
  10. public class All_BConsumer {
  12.     public static void main(String[] argv) throws IOException,
  13.             InterruptedException, TimeoutException {
  14.         ConnectionFactory factory = new ConnectionFactory();
  15.         //连接rabbitMq的地址
  16.         factory.setHost("zjj101");
  17.         /*设置登陆用户名和密码*/
  18.         factory.setUsername("guest");
  19.         factory.setPassword("guest");
  20.         /*设置虚拟主机地址,不能设置错了,不然会报错误*/
  21.         factory.setVirtualHost("/");
  22.         // 打开连接和创建频道，与发送端一样
  23.         Connection connection = factory.newConnection();
  24.         final Channel channel = connection.createChannel();
  26.         channel.exchangeDeclare(TopicProducer.EXCHANGE_NAME,
  27.                 BuiltinExchangeType.TOPIC);
  28.         // 声明一个随机队列
  29.         String queueName = channel.queueDeclare().getQueue();
  30.         //前面怎么样的无所谓,最后一个是B章节就可以了,
  31.         //#匹配无数个
  32.         channel.queueBind(queueName, TopicProducer.EXCHANGE_NAME, "#.B");
  34.         System.out.println(" [*] Waiting for messages:");
  36.         // 创建队列消费者
  37.         final Consumer consumerA = new DefaultConsumer(channel) {
  38.             @Override
  39.             public void handleDelivery(String consumerTag,
  40.                                        Envelope envelope,
  41.                                        AMQP.BasicProperties properties,
  42.                                        byte[] body)
  43.                     throws IOException {
  44.                 String message = new String(body, "UTF-8");
  45.                 System.out.println(" AllConsumer Received "
  46.                         + envelope.getRoutingKey()
  47.                         + "':'" + message + "'");
  48.             }
  49.         };
  50.         channel.basicConsume(queueName, true, consumerA);
  51.     }
  52. }

启动完这一个消费者再启动生产者后输出结果:

所有结尾为b的 routingKey都被消费了

  1.  [*] Waiting for messages:
  2.  AllConsumer Received king.kafka.B':'Hello Topic_[0,0,1]'
  3.  AllConsumer Received king.jvm.B':'Hello Topic_[0,1,1]'
  4.  AllConsumer Received king.redis.B':'Hello Topic_[0,2,1]'
  5.  AllConsumer Received mark.kafka.B':'Hello Topic_[1,0,1]'
  6.  AllConsumer Received mark.jvm.B':'Hello Topic_[1,1,1]'
  7.  AllConsumer Received mark.redis.B':'Hello Topic_[1,2,1]'
  8.  AllConsumer Received james.kafka.B':'Hello Topic_[2,0,1]'
  9.  AllConsumer Received james.jvm.B':'Hello Topic_[2,1,1]'
  10.  AllConsumer Received james.redis.B':'Hello Topic_[2,2,1]'

消费者 -RoutingKey写全了

  1. import com.rabbitmq.client.*;
  2. import java.io.IOException;
  3. import java.util.concurrent.TimeoutException;
  5. /**
  6.  * 6、关注king老师kafka的A章节，怎么办？
  7.  *类说明：
  8.  */
  9. public class King_kafka_AConsumer {
  11.     public static void main(String[] argv) throws IOException,
  12.             InterruptedException, TimeoutException {
  13.         ConnectionFactory factory = new ConnectionFactory();
  14.         //连接rabbitMq的地址
  15.         factory.setHost("zjj101");
  16.         /*设置登陆用户名和密码*/
  17.         factory.setUsername("guest");
  18.         factory.setPassword("guest");
  19.         /*设置虚拟主机地址,不能设置错了,不然会报错误*/
  20.         factory.setVirtualHost("/");
  22.         // 打开连接和创建频道，与发送端一样
  23.         Connection connection = factory.newConnection();
  24.         final Channel channel = connection.createChannel();
  26.         channel.exchangeDeclare(TopicProducer.EXCHANGE_NAME,
  27.                 BuiltinExchangeType.TOPIC);
  28.         // 声明一个随机队列
  29.         String queueName = channel.queueDeclare().getQueue();
  31.         channel.queueBind(queueName,TopicProducer.EXCHANGE_NAME, "king.kafka.A");
  33.         System.out.println(" [*] Waiting for messages:");
  35.         // 创建队列消费者
  36.         final Consumer consumerA = new DefaultConsumer(channel) {
  37.             @Override
  38.             public void handleDelivery(String consumerTag,
  39.                                        Envelope envelope,
  40.                                        AMQP.BasicProperties properties,
  41.                                        byte[] body)
  42.                     throws IOException {
  43.                 String message = new String(body, "UTF-8");
  44.                 System.out.println(" AllConsumer Received "
  45.                         + envelope.getRoutingKey()
  46.                         + "':'" + message + "'");
  47.             }
  48.         };
  49.         channel.basicConsume(queueName, true, consumerA);
  50.     }
  51. }

启动完这一个消费者再启动生产者后输出结果:

只有routingKey为king.kafka.A的 routingKey都被消费了

  1.  [*] Waiting for messages:
  2.  AllConsumer Received king.kafka.A':'Hello Topic_[0,0,0]'