在之前的Work queues中，我们将消息发送给多个消费者，但是同一条消息只被A or B处理，假设有这样一个场景：用户注册后，需要向用户发短信，还需要发邮件，也就是同一条消息需要被A and B处理，这就需要用到Publish/Subscribe模式：
1.Exchanges
交换器，就是上图中的X。在RabbitMQ中，生产者的消息都不会直接发送到队列，而是会发到交换器，交换器接收生产者的消息，然后将消息推送给队列，起到承上启下的作用。交换器收到消息后，具体要如何处理，是推送给某一个队列？某几个队列？丢弃？这就要通过交换器类型来决定。要实现文章开头的那个需求，我们可以将类型定义成fanout（呈扇形展开），交换器将会把消息广播到所有队列。
声明交换器：
channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
在之前Work queues中，我们是这样发送消息的:
channel.basicPublish("", "hello", null, message.getBytes());
第一个参数就是指定交换器，""表示默认交换器，第二个参数为routingKey，表示交换器要把消息推给routingKey标识的队列，即生产者将消息发送到默认交换器，交换器将消息推送给名字为hello的队列。
现在我们将这样发送消息：
channel.basicPublish( EXCHANGE_NAME, "", null, message.getBytes());
生产者将消息发送给上面声明的交换器EXCHANGE_NAME，交换器将消息推送给所有队列，所以这里routingKey使用""，因为我们是广播，不需要指明routingKey。
2. 队列绑定到交换器
在之前声明队列时，我们指定了队列名称，正是队列名称匹配了routingKey，才能保证消息到达指定队列：
channel.queueDeclare(QUEUE_NAME, true, false, false, null);
但现在我们要发送给所有队列，所以队列名不需要和routingKey匹配了，可以随便起个名字。
在声明队列之后，我们还需要将队列和交换器绑定，这样交换器才能将消息推送给绑定的队列，如果交换器没有绑定队列，交换器收到消息后将直接丢弃。
//声明队列
channel.queueDeclare(QUEUE_NAME,false,false,false,null);
//绑定队列到交换器
channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "");
3.完整代码
生产者
//发布订阅模式生产者
public class Send {
public static final String EXCHANGE_NAME = "exchange_fanout";
public static void main(String[] args) throws IOException, TimeoutException {
Connection connection = ConnectionUtils.getConnection();
Channel channel = connection.createChannel();
//声明交换器fanout类似于广播
channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.FANOUT);
String msg = "hello publish/subscribe";
//向交换机发送消息，由于交换机没有存储消息的能力，所以如果没有队列绑定到交换机，消息将被丢弃
channel.basicPublish(EXCHANGE_NAME, "", null, msg.getBytes());
System.out.println("向交换机发送了消息"+msg);
channel.close();
connection.close();
}
}
消费者1
/**发布订阅模式，消费者**/
public class Receive1 {
public static final String QUEUE_NAME = "psQueue_email";
public static final String EXCHANGE_NAME = "exchange_fanout";
public static void main(String[] args) throws Exception{
Connection connection = ConnectionUtils.getConnection();
final Channel channel = connection.createChannel();
//声明队列
channel.queueDeclare(QUEUE_NAME,false,false,false,null);
//绑定队列到转发器
channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "");
//RabbitMQ在收到我反馈的ack后才投递下一条消息给我
channel.basicQos(1);
DefaultConsumer consumer = new DefaultConsumer(channel){
//收到消息就会触发
@Override
public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body)
throws IOException {
String msg = new String(body,"utf-8");
System.out.println("消费者1:"+msg);
try {
Thread.sleep(2000);
} catch (InterruptedException e) {
e.printStackTrace();
}
//向生产者发送回执消息
channel.basicAck(envelope.getDeliveryTag(), false);
}
};
//关闭自动应答
boolean autoAck = false;
//监听队列
channel.basicConsume(QUEUE_NAME, autoAck, consumer);
}
}
消费者2
/**发布订阅模式，消费者**/
public class Receive2 {
public static final String QUEUE_NAME = "psQueue_msm";
public static final String EXCHANGE_NAME = "exchange_fanout";
public static void main(String[] args) throws Exception{
Connection connection = ConnectionUtils.getConnection();
final Channel channel = connection.createChannel();
//声明队列
channel.queueDeclare(QUEUE_NAME,false,false,false,null);
//绑定队列到转发器
channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "");
//RabbitMQ在收到我反馈的ack后才投递下一条消息给我
channel.basicQos(1);
DefaultConsumer consumer = new DefaultConsumer(channel){
//收到消息就会触发
@Override
public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body)
throws IOException {
String msg = new String(body,"utf-8");
System.out.println("消费者2:"+msg);
try {
Thread.sleep(2000);
} catch (InterruptedException e) {
e.printStackTrace();
}
//向生产者发送回执消息
channel.basicAck(envelope.getDeliveryTag(), false);
}
};
//关闭自动应答
boolean autoAck = false;
//监听队列
channel.basicConsume(QUEUE_NAME, autoAck, consumer);
}
}
