消息中间件(五)-----AMQP概论、rabbitMQ入门
AMQP概论
AMQP
是应用层协议的一个开放标准,为面向消息的中间件设计。基于此协议的客户端与消息中间件可传递消息，并不受客户端/中间件不同产品，不同的开发语言等条件的限制。目标是实现一种在全行业广泛使用的标准消息中间件技术，以便降低企业和系统集成的开销，并且向大众提供工业级的集成服务。主要实现有 RabbitMQ。
包括的要素
生产者、消费者、消息
生产者:消息的创建者，发送到rabbitmq；
消费者：连接到rabbitmq，订阅到队列上，消费消息，持续订阅(basicConsumer)和单条订阅(basicGet).
消息：包含有效载荷和标签，有效载荷指要传输的数据，，标签描述了有效载荷，并且rabbitmq用它来决定谁获得消息，消费者只能拿到有效载荷，并不知道生产者是谁。
信道
信道，概念：信道是生产者、消费者与rabbit通信的渠道，生产者publish或是消费者subscribe一个队列都是通过信道来通信的。信道是建立在TCP连接上的虚拟连接，什么意思呢？就是说rabbitmq在一条TCP上建立成百上千个信道来达到多个线程处理，这个TCP被多个线程共享，每个线程对应一个信道，信道在rabbit都有唯一的ID ,保证了信道私有性，对应上唯一的线程使用。
疑问：为什么不建立多个TCP连接呢？原因是rabbit保证性能，系统为每个线程开辟一个TCP是非常消耗性能，每秒成百上千的建立销毁TCP会严重消耗系统。所以rabbitmq选择建立多个信道（建立在tcp的虚拟连接）连接到rabbit上。
理解：可以理解为登录qq，客户端与qq服务器只有一个连接，但是qq可以打开多个窗口与多个好友聊天，每个窗口可以理解为是一个信道。
交换器、队列、绑定、路由键
队列通过路由键（routing key，某种确定的规则）绑定到交换器，生产者将消息发布到交换器，交换器根据绑定的路由键将消息路由到特定队列，然后由订阅这个队列的消费者进行接收。
常见问题
如果消息达到无人订阅的队列会怎么办？消息会一直在队列中等待，RabbitMq默认队列是无限长度的。
多个消费者订阅到同一队列怎么办？消息以循环的方式发送给消费者，每个消息只会发送给一个消费者。
消息路由到了不存在的队列怎么办？一般情况下，凉拌，RabbitMq会忽略，当这个消息不存在，也就是这消息丢了。
消息的确认
消费者收到的每一条消息都必须进行确认（自动确认和自行确认）。
消费者在声明队列时，可以指定autoAck参数，当autoAck=false时，RabbitMQ会等待消费者显式发回ack信号后才从内存(和磁盘，如果是持久化消息的话)中移去消息。否则，RabbitMQ会在队列中消息被消费后立即删除它。
采用消息确认机制后，只要令autoAck=false，消费者就有足够的时间处理消息(任务)，不用担心处理消息过程中消费者进程挂掉后消息丢失的问题，因为RabbitMQ会一直持有消息直到消费者显式调用basicAck为止。
当autoAck=false时，对于RabbitMQ服务器端而言，队列中的消息分成了两部分：一部分是等待投递给消费者的消息；一部分是已经投递给消费者，但是还没有收到消费者ack信号的消息。如果服务器端一直没有收到消费者的ack信号，并且消费此消息的消费者已经断开连接，则服务器端会安排该消息重新进入队列，等待投递给下一个消费者（也可能还是原来的那个消费者）。
RabbitMQ不会为未ack的消息设置超时时间，它判断此消息是否需要重新投递给消费者的唯一依据是消费该消息的消费者连接是否已经断开。这么设计的原因是RabbitMQ允许消费者消费一条消息的时间可以很久很久。
交换器类型
共有四种direct,fanout,topic,headers，其种headers(几乎和direct一样)不实用，可以忽略。
Direct
路由键完全匹配，消息被投递到对应的队列，每个amqp的实现都必须有一个direct交换器，包含一个空白字符串名称的默认交换器。声明一个队列时，会自动绑定到默认交换器，并且以队列名称作为路由键：channel->basic_public($msg,’ ’,’queue-name’)
Fanout
消息广播到绑定的队列
Topic
通过使用“*”和“#”，使来自不同源头的消息到达同一个队列，”.”将路由键分为了几个标识符，“*”匹配1个，“#”匹配一个或多个。例如日志处理：
假设有交换器log-exchange，
日志级别有error,info,warning，
应用模块有user,order,email，
服务器有 A、B、C、D
路由键的规则为 服务器+“.”+日志级别+“.”+应用模块名，如：A. info .email。
1、要关注A服务器发送的所有应用错误的消息，怎么做？
声明队列名称为“a-app-error-queue”并绑定到交换器上：channel. queueBind (‘a-app-error-queue’,’logs-change’,’A.error.*’)
2、关注B服务器发送的的所有日志，怎么办？
声明队列名称为“b-all-queue”并绑定到交换器上：channel. queueBind (b-all-queue’,’logs-change’,’ B.#’)或channel. queueBind (b-all-queue’,’logs-change’,’ B.*.*’)
3、关注所有服务器发送的email的所有日志，怎么办？
声明队列名称为“email-all-queue”并绑定到交换器上：channel. queueBind (email -all-queue’,’logs-change’,’ *.*.emal’)
4、想要接收所有日志：channel->queue_bind(‘all-log’,’logs-change’,’#’)
虚拟主机
虚拟消息服务器，vhost，本质上就是一个mini版的mq服务器，有自己的队列、交换器和绑定，最重要的，自己的权限机制。Vhost提供了逻辑上的分离，可以将众多客户端进行区分，又可以避免队列和交换器的命名冲突。Vhost必须在连接时指定，rabbitmq包含缺省vhost：“/”，通过缺省用户和口令guest进行访问。
rabbitmq里创建用户，必须要被指派给至少一个vhost，并且只能访问被指派内的队列、交换器和绑定。Vhost必须通过rabbitmq的管理控制工具创建。
RabbitMQ在Windows下安装和运行
1、下载Erlang：
http://www.erlang.org/downloads/19.2
2、下载Windows版RabbitMq：
http://www.rabbitmq.com/releases/rabbitmq-server/v3.6.6/rabbitmq-server-3.6.6.exe
3、安装并配置环境变量：
增加变量ERLANG_HOME C:\Program Files\erl8.2
path下添加 %ERLANG_HOME%\bin，如
增加变量RABBITMQ_BASE C:\Program Files\RabbitMQ Server\rabbitmq_server-3.6.6
path下添加 %RABBITMQ_BASE%\sbin;%RABBITMQ_BASE%\ebin
4、在开始菜单中启动服务
5、可以在安装目录的sbin下运行rabbitmqctl.bat status检测是否安装成功
原生Java客户端进行消息通信
Direct
DirectProducer：direct类型交换器的生产者
NormalConsumer：普通的消费者
MulitBindConsumer：队列绑定到交换器上时，是允许绑定多个路由键的，也就是多重绑定
MulitChannelConsumer：一个连接下允许有多个信道
MulitConsumerOneQueue：一个队列多个消费者，则会表现出消息在消费者之间的轮询发送。
Fanout
消息广播到绑定的队列
通过测试表明，不管我们如何调整生产者和消费者的路由键，都对消息的接受没有影响。
Topic
通过使用“*”和“#”，使来自不同源头的消息到达同一个队列，”.”将路由键分为了几个标识符，“*”匹配1个，“#”匹配一个或多个。例如日志处理：
假设有交换器log-exchange，日志级别有error,info,warning，应用模块有user,order,email，服务器有 A、B、C、D
路由键的规则为 服务器+“.”+日志级别+“.”+应用模块名，如：A. info .email。
1、要关注A服务器发送的所有应用错误的消息，怎么做？
声明队列名称为“a-app-error-queue”并绑定到交换器上：channel. queueBind (‘a-app-error-queue’,’logs-change’,’A.error.*’)
2、关注B服务器发送的的所有日志，怎么办？
声明队列名称为“b-all-queue”并绑定到交换器上：channel. queueBind (b-all-queue’,’logs-change’,’ B.#’)或channel. queueBind (b-all-queue’,’logs-change’,’ B.*.*’)
3、关注所有服务器发送的email的所有日志，怎么办？
声明队列名称为“email-all-queue”并绑定到交换器上：channel. queueBind (email -all-queue’,’logs-change’,’ *.*.email’)
4、想要接收所有日志：channel->queue_bind(‘all-log’,’logs-change’,’#’)
示例
引入依赖
客户端需要amqp-client-5.0.0.jar和slf4j-api-1.6.1.jar
<dependency> <groupId>com.rabbitmq</groupId> <artifactId>amqp-client</artifactId> <version>5.0.0</version> </dependency>
注意：5系列的版本最好使用JDK8及以上， 低于JDK8可以使用4.x(具体的版本号到Maven的中央仓库查)的版本。
代码
1、direct-----消费的多种场景
DirectProducer
package cn.enjoyedu.exchange.direct;
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;
/**
* 类说明：direct类型交换器的生产者
*/
public class DirectProducer {
public final static String EXCHANGE_NAME = "direct_logs";
public static void main(String[] args) throws IOException, TimeoutException {
/* 创建连接,连接到RabbitMQ*/
ConnectionFactory connectionFactory = new ConnectionFactory();
connectionFactory.setHost("127.0.0.1");
Connection connection = connectionFactory.newConnection();
/*创建信道*/
Channel channel = connection.createChannel();
/*创建交换器*/
channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT);
/*日志消息级别，作为路由键使用*/
String[] serverities = {"error", "info", "warning"};
for (int i = 0; i < 3; i++) {
String severity = serverities[i % 3];
String msg = "Hellol,RabbitMq" + (i + 1);
/*发布消息，需要参数：交换器，路由键，其中以日志消息级别为路由键*/
channel.basicPublish(EXCHANGE_NAME, severity, null, msg.getBytes());
System.out.println("Send " + severity + ":" + msg);
}
channel.close();
connection.close();
}
}NormalConsumer
package cn.enjoyedu.exchange.direct;
import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
/**
* 类说明：普通的消费者
*/
public class NormalConsumer {
public static void main(String[] argv) throws IOException, TimeoutException {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("127.0.0.1");
// 打开连接和创建频道，与发送端一样
Connection connection = factory.newConnection();
final Channel channel = connection.createChannel();
channel.exchangeDeclare(DirectProducer.EXCHANGE_NAME,"direct");
/*声明一个队列*/
String queueName = "focuserror";
channel.queueDeclare(queueName, false, false,false,null);
/*绑定，将队列和交换器通过路由键进行绑定*/
String routekey = "error";/*表示只关注error级别的日志消息*/
channel.queueBind(queueName, DirectProducer.EXCHANGE_NAME, routekey);
System.out.println("waiting for message........");
/*声明了一个消费者*/
final Consumer consumer = new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
String message = new String(body, "UTF-8");
System.out.println("Received[" + envelope.getRoutingKey() + "]" + message);
}
};
/*消费者正式开始在指定队列上消费消息*/
channel.basicConsume(queueName, true, consumer);
}
}运行结果：只能接收到路由键为error的消息
MulitBindConsumer
package cn.enjoyedu.exchange.direct;
import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
/**
*类说明：队列和交换器的多重绑定
*/
public class MulitBindConsumer {
public static void main(String[] argv) throws IOException, InterruptedException, TimeoutException {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("127.0.0.1");
// 打开连接和创建频道，与发送端一样
Connection connection = factory.newConnection();
final Channel channel = connection.createChannel();
channel.exchangeDeclare(DirectProducer.EXCHANGE_NAME, "direct");
//声明一个随机队列
String queueName = channel.queueDeclare().getQueue();
/*队列绑定到交换器上时，是允许绑定多个路由键的，也就是多重绑定*/
String[] severities={"error","info","warning"};
for(String serverity:severities){
channel.queueBind(queueName,DirectProducer.EXCHANGE_NAME, serverity);
}
System.out.println(" [*] Waiting for messages:");
// 创建队列消费者
final Consumer consumerA = new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
String message = new String(body, "UTF-8");
System.out.println(" Received " + envelope.getRoutingKey() + ":‘" + message + "‘");
}
};
channel.basicConsume(queueName, true, consumerA);
}
}运行结果：能接收到路由键为error、info、warning所有的消息
MulitChannelConsumer
package cn.enjoyedu.exchange.direct;
import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
/**
* 类说明：一个连接多个信道
*/
public class MulitChannelConsumer {
private static class ConsumerWorker implements Runnable {
final Connection connection;
public ConsumerWorker(Connection connection) {
this.connection = connection;
}
public void run() {
try {
/*创建一个信道，意味着每个线程单独一个信道*/
final Channel channel = connection.createChannel();
channel.exchangeDeclare(DirectProducer.EXCHANGE_NAME,"direct");
// 声明一个随机队列
String queueName = channel.queueDeclare().getQueue();
//消费者名字，打印输出用
final String consumerName = Thread.currentThread().getName() + "-all";
//所有日志严重性级别
String[] severities = {"error", "info", "warning"};
for (String severity : severities) {
//关注所有级别的日志（多重绑定）
channel.queueBind(queueName, DirectProducer.EXCHANGE_NAME, severity);
}
System.out.println("[" + consumerName + "] Waiting for messages:");
// 创建队列消费者
final Consumer consumerA = new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag,
Envelope envelope,
AMQP.BasicProperties
properties,
byte[] body)
throws IOException {
String message = new String(body, "UTF-8");
System.out.println(consumerName + " Received " + envelope.getRoutingKey() + ":‘" + message + "‘");
}
};
channel.basicConsume(queueName, true, consumerA);
} catch (Exception e) {
e.printStackTrace();
}
}
}
public static void main(String[] argv) throws IOException, InterruptedException, TimeoutException {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("127.0.0.1");
// 打开连接和创建频道，与发送端一样
Connection connection = factory.newConnection();
//一个连接多个信道
for (int i = 0; i < 2; i++) {
/*将连接作为参数，传递给每个线程*/
Thread worker = new Thread(new ConsumerWorker(connection));
worker.start();
}
}
}运行结果：能接收到路由键为error、info、warning所有的消息
MulitConsumerOneQueue
package cn.enjoyedu.exchange.direct;
import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
/**
*类说明：一个队列多个消费者，则会表现出消息在消费者之间的轮询发送。
*/
public class MulitConsumerOneQueue {
private static class ConsumerWorker implements Runnable{
final Connection connection;
final String queueName;
public ConsumerWorker(Connection connection,String queueName) {
this.connection = connection;
this.queueName = queueName;
}
public void run() {
try {
/*创建一个信道，意味着每个线程单独一个信道*/
final Channel channel = connection.createChannel();
channel.exchangeDeclare(DirectProducer.EXCHANGE_NAME,
"direct");
/*声明一个队列,rabbitmq，如果队列已存在，不会重复创建*/
channel.queueDeclare(queueName,
false,false,
false,null);
//消费者名字，打印输出用
final String consumerName
= Thread.currentThread().getName();
//所有日志严重性级别
String[] severities={"error","info","warning"};
for (String severity : severities) {
//关注所有级别的日志（多重绑定）
channel.queueBind(queueName,
DirectProducer.EXCHANGE_NAME, severity);
}
System.out.println(" ["+consumerName+"] Waiting for messages:");
// 创建队列消费者
final Consumer consumerA = new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag,
Envelope envelope,
AMQP.BasicProperties
properties,
byte[] body)
throws IOException {
String message =
new String(body, "UTF-8");
System.out.println(consumerName
+" Received " + envelope.getRoutingKey()
+ ":‘" + message + "‘");
}
};
channel.basicConsume(queueName, true, consumerA);
} catch (Exception e) {
e.printStackTrace();
}
}
}
public static void main(String[] argv) throws IOException,
InterruptedException, TimeoutException {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("127.0.0.1");
// 打开连接和创建频道，与发送端一样
Connection connection = factory.newConnection();
//3个线程，线程之间共享队列,一个队列多个消费者
String queueName = "focusAll";
for(int i=0;i<3;i++){
/*将队列名作为参数，传递给每个线程*/
Thread worker =new Thread(new ConsumerWorker(connection,queueName));
worker.start();
}
}
}运行结果：能接收到路由键为error、info、warning所有的消息，不过消息由多个消费者循环消费。
2、fanout
fanout就类似于广播，不论路由键怎么设置，都会广播到每个队列上。
3、topic
TopicProducer
package cn.enjoyedu.exchange.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;
public class TopicProducer {
public final static String EXCHANGE_NAME = "topic_logs";
public static void main(String[] args) throws IOException, TimeoutException {
/**
* 创建连接连接到RabbitMQ
*/
ConnectionFactory factory = new ConnectionFactory();
// 设置MabbitMQ所在主机ip或者主机名
factory.setHost("127.0.0.1");
// 创建一个连接
Connection connection = factory.newConnection();
// 创建一个信道
Channel channel = connection.createChannel();
// 指定转发
channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.TOPIC);
/*日志消息，路由键最终格式类似于：info.order.B*/
String[] severities={"error","info","warning"};
for(int i=0;i<3;i++){
String[] modules={"user","order","email"};
for(int j=0;j<3;j++){
String[] servers={"A","B","C"};
for(int k=0;k<3;k++){
// 发送的消息
String message = "Hello Topic_["+i+","+j+","+k+"]";
String routeKey = severities[i%3]+"."+modules[j%3] +"."+servers[k%3];
channel.basicPublish(EXCHANGE_NAME,routeKey, null, message.getBytes());
System.out.println(" [x] Sent ‘" + routeKey +":‘" + message + "‘");
}
}
}
// 关闭频道和连接
channel.close();
connection.close();
}
}AllConsumer
package cn.enjoyedu.exchange.topic;
import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
public class AllConsumer {
public static void main(String[] argv) throws IOException, InterruptedException, TimeoutException {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("127.0.0.1");
// 打开连接和创建频道，与发送端一样
Connection connection = factory.newConnection();
final Channel channel = connection.createChannel();
channel.exchangeDeclare(TopicProducer.EXCHANGE_NAME, BuiltinExchangeType.TOPIC);
// 声明一个随机队列
String queueName = channel.queueDeclare().getQueue();
channel.queueBind(queueName,TopicProducer.EXCHANGE_NAME, "#");
System.out.println(" [*] Waiting for messages:");
// 创建队列消费者
final Consumer consumerA = new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
String message = new String(body, "UTF-8");
System.out.println(" AllConsumer Received " + envelope.getRoutingKey() + "‘:‘" + message + "‘");
}
};
channel.basicConsume(queueName, true, consumerA);
}
}运行结果：一共收到27条消息。
EmailAllConsumer
package cn.enjoyedu.exchange.topic;
import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
public class EmailAllConsumer {
public static void main(String[] argv) throws IOException, InterruptedException, TimeoutException {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("127.0.0.1");
// 打开连接和创建频道，与发送端一样
Connection connection = factory.newConnection();
final Channel channel = connection.createChannel();
channel.exchangeDeclare(TopicProducer.EXCHANGE_NAME, BuiltinExchangeType.TOPIC);
// 声明一个随机队列
String queueName = channel.queueDeclare().getQueue();
channel.queueBind(queueName,TopicProducer.EXCHANGE_NAME, "*.email.*");
System.out.println(" [*] Waiting for messages:");
// 创建队列消费者
final Consumer consumerA = new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
String message = new String(body, "UTF-8");
System.out.println(" AllConsumer Received " + envelope.getRoutingKey() + "‘:‘" + message + "‘");
}
};
channel.basicConsume(queueName, true, consumerA);
}
}运行结果：
ErrorAllConsumer
package cn.enjoyedu.exchange.topic;
import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
public class ErrorAllConsumer {
public static void main(String[] argv) throws IOException, InterruptedException, TimeoutException {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("127.0.0.1");
// 打开连接和创建频道，与发送端一样
Connection connection = factory.newConnection();
final Channel channel = connection.createChannel();
channel.exchangeDeclare(TopicProducer.EXCHANGE_NAME, BuiltinExchangeType.TOPIC);
// 声明一个随机队列
String queueName = channel.queueDeclare().getQueue();
channel.queueBind(queueName,TopicProducer.EXCHANGE_NAME, "error.#");
System.out.println(" [*] Waiting for messages:");
// 创建队列消费者
final Consumer consumerA = new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
String message = new String(body, "UTF-8");
System.out.println(" AllConsumer Received " + envelope.getRoutingKey() + "‘:‘" + message + "‘");
}
};
channel.basicConsume(queueName, true, consumerA);
}
}运行结果：
ServerAErrorConsumer
package cn.enjoyedu.exchange.topic;
import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
public class ServerAErrorConsumer {
public static void main(String[] argv) throws IOException, InterruptedException, TimeoutException {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("127.0.0.1");
// 打开连接和创建频道，与发送端一样
Connection connection = factory.newConnection();
final Channel channel = connection.createChannel();
channel.exchangeDeclare(TopicProducer.EXCHANGE_NAME, BuiltinExchangeType.TOPIC);
// 声明一个随机队列
String queueName = channel.queueDeclare().getQueue();
channel.queueBind(queueName,TopicProducer.EXCHANGE_NAME, "error.*.A");
System.out.println(" [*] Waiting for messages:");
// 创建队列消费者
final Consumer consumerA = new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
String message = new String(body, "UTF-8");
System.out.println(" AllConsumer Received " + envelope.getRoutingKey() + "‘:‘" + message + "‘");
}
};
channel.basicConsume(queueName, true, consumerA);
}
}运行结果：
ServerBAllConsumer
package cn.enjoyedu.exchange.topic;
import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
public class ServerBAllConsumer {
public static void main(String[] argv) throws IOException, InterruptedException, TimeoutException {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("127.0.0.1");
// 打开连接和创建频道，与发送端一样
Connection connection = factory.newConnection();
final Channel channel = connection.createChannel();
channel.exchangeDeclare(TopicProducer.EXCHANGE_NAME, BuiltinExchangeType.TOPIC);
// 声明一个随机队列
String queueName = channel.queueDeclare().getQueue();
channel.queueBind(queueName,TopicProducer.EXCHANGE_NAME, "#.B");
System.out.println(" [*] Waiting for messages:");
// 创建队列消费者
final Consumer consumerA = new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
String message = new String(body, "UTF-8");
System.out.println(" AllConsumer Received " + envelope.getRoutingKey() + "‘:‘" + message + "‘");
}
};
channel.basicConsume(queueName, true, consumerA);
}
}运行结果：
