RabbitMQJava系列4-Publish/Subscribe
Posted zytiger
tags:
篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了RabbitMQJava系列4-Publish/Subscribe相关的知识,希望对你有一定的参考价值。
发布订阅模式
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的主要内容,如果未能解决你的问题,请参考以下文章