發布訂閱模式

X：交換機（轉發器）

生產者把消息發送到交換機，交換機把消息發送到隊列中，隊列需要綁定到交換機。

1，一個生產者，多個消費者

2，每個消費者都有自己的隊列

Java代碼實現

生產者

import java.io.IOException;
import java.util.concurrent.TimeoutException;

import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.zy.rabbitmq.util.ConnectionUtil;

/**
* 發布/訂閱模式
* 消息生產者
*
* @author zy
*
*/
public class Send {
private static final String EXCHAGNE_NAME = "test_exchange_fanot";

public static void main(String[] args) throws IOException, TimeoutException {
// 獲取連接
Connection connection = ConnectionUtil.getConnection();
// 從連接中獲取一個通道
Channel channel = connection.createChannel();
// 交換機聲明
channel.exchangeDeclare(EXCHAGNE_NAME, BuiltinExchangeType.FANOUT);
String msg = "發布/訂閱消息!";
// 發送
channel.basicPublish(EXCHAGNE_NAME, "", null, msg.getBytes());
System.out.println("send msg:" + msg);
channel.close();
connection.close();
}
}

消費者1

import java.io.IOException;
import java.util.concurrent.TimeoutException;

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.Consumer;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;
import com.zy.rabbitmq.util.ConnectionUtil;

/**
* 消費者消費消息
*
* @author zy
*
*/
public class Receive {

private static final String QUEUE_NAME = "test_queue_fanout_email";

private static final String EXCHAGNE_NAME = "test_exchange_fanot";

public static void main(String[] args) throws IOException, TimeoutException {
// 獲取連接
Connection connection = ConnectionUtil.getConnection();
// 從連接中獲取一個通道
final Channel channel = connection.createChannel();
// 隊列聲明
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
// 綁定隊列到交換機
channel.queueBind(QUEUE_NAME, EXCHAGNE_NAME, "");
channel.basicQos(1);
// 基於事件
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(" receive email msg:" + message);
channel.basicAck(envelope.getDeliveryTag(), false);
}
};

// 監聽隊列
channel.basicConsume(QUEUE_NAME, false, consumer);
}

}

消費者2

import java.io.IOException;
import java.util.concurrent.TimeoutException;

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.Consumer;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;
import com.zy.rabbitmq.util.ConnectionUtil;

/**
* 消費者消費消息
*
* @author zy
*
*/
public class Receive2 {

private static final String QUEUE_NAME = "test_queue_fanout_sms";

private static final String EXCHAGNE_NAME = "test_exchange_fanot";

public static void main(String[] args) throws IOException, TimeoutException {
// 獲取連接
Connection connection = ConnectionUtil.getConnection();
// 從連接中獲取一個通道
final Channel channel = connection.createChannel();
// 隊列聲明
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
// 綁定隊列到交換機
channel.queueBind(QUEUE_NAME, EXCHAGNE_NAME, "");
channel.basicQos(1);
// 基於事件
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(" receive sms msg:" + message);
channel.basicAck(envelope.getDeliveryTag(), false);
}
};

// 監聽隊列
channel.basicConsume(QUEUE_NAME, false, consumer);
}

}

總結：發布/訂閱模式需要隊列綁定到交換機，不定義路由鍵

參考：http://www.rabbitmq.com/tutorials/tutorial-four-java.html

RabbitMQJava系列4-Publish/Subscribe