（转） RabbitMQ学习之发布/订阅（java）

http://blog.csdn.net/zhu_tianwei/article/details/40887733

参考：http://blog.csdn.NET/lmj623565791/article/details/37657225

模拟发布订阅模式，一个消息发给多个消费者。实现一个发送日志，一个接收者将接收到的数据写到硬盘上，与此同时，另一个接收者把接收到的消息展现在屏幕上。

转发器类型使用：fanout。fanout类型转发器特别简单，把所有它介绍到的消息，广播到所有它所知道的队列。

channel.exchangeDeclare("logs","fanout");

1.发送日志SendLog.Java

[java] view plain copy

print?

package cn.slimsmart.rabbitmq.demo.fanout;

import java.util.Date;

import com.rabbitmq.client.AMQP;

import com.rabbitmq.client.Channel;

import com.rabbitmq.client.Connection;

import com.rabbitmq.client.ConnectionFactory;

public class SendLog {

// 转发器

private final static String EXCHANGE_NAME = "ex_log";

@SuppressWarnings("deprecation")

public static void main(String[] args) throws Exception {

// 创建连接和频道

ConnectionFactory factory = new ConnectionFactory();

factory.setHost("192.168.101.174");

// 指定用户 密码

factory.setUsername("admin");

factory.setPassword("admin");

// 指定端口

factory.setPort(AMQP.PROTOCOL.PORT);

Connection connection = factory.newConnection();

Channel channel = connection.createChannel();

/**

* 声明转发器和类型 可用的转发器类型Direct Topic Headers Fanout 参考：http://melin.iteye.com/blog/691265

* Direct Exchange – 处理路由键。需要将一个队列绑定到交换机上，要求该消息与一个特定的路由键完全匹配。

* Fanout Exchange – 不处理路由键。你只需要简单的将队列绑定到交换机上。一个发送到交换机的消息都会被转发到与该交换机绑定的所有队列上。

* 很像子网广播，每台子网内的主机都获得了一份复制的消息。Fanout交换机转发消息是最快的。

* Topic Exchange – 将路由键和某模式进行匹配。此时队列需要绑定要一个模式上。符号“#”匹配一个或多个词，符号“*”匹配不多不少一个词。

* 因此“audit.#”能够匹配到“audit.irs.corporate”，但是“audit.*” 只会匹配到“audit.irs”。

*/

channel.exchangeDeclare(EXCHANGE_NAME, "fanout");

String message = new Date().toLocaleString() + " : log something";

// 指定消息发送到的转发器

channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes());

System.out.println(" [x] Sent '" + message + "'");

channel.close();

connection.close();

}

}

2.接受写文件类代码ReceiveLogsToFile.java

创建一个队列，然后将队列与转发器绑定，然后将消费者与该队列绑定，然后写入日志文件。

[java] view plain copy

print?

package cn.slimsmart.rabbitmq.demo.fanout;

import java.io.File;

import java.io.FileNotFoundException;

import java.io.FileOutputStream;

import java.io.IOException;

import java.text.SimpleDateFormat;

import java.util.Date;

import com.rabbitmq.client.AMQP;

import com.rabbitmq.client.Channel;

import com.rabbitmq.client.Connection;

import com.rabbitmq.client.ConnectionFactory;

import com.rabbitmq.client.QueueingConsumer;

public class ReceiveLogsToFile {

private final static String EXCHANGE_NAME = "ex_log";

public static void main(String[] args) throws Exception {

// 创建连接和频道

ConnectionFactory factory = new ConnectionFactory();

factory.setHost("192.168.101.174");

// 指定用户 密码

factory.setUsername("admin");

factory.setPassword("admin");

// 指定端口

factory.setPort(AMQP.PROTOCOL.PORT);

Connection connection = factory.newConnection();

Channel channel = connection.createChannel();

/**

* 声明转发器和类型 可用的转发器类型Direct Topic Headers Fanout

* Direct Exchange – 处理路由键。需要将一个队列绑定到交换机上，要求该消息与一个特定的路由键完全匹配。

* Fanout Exchange – 不处理路由键。你只需要简单的将队列绑定到交换机上。一个发送到交换机的消息都会被转发到与该交换机绑定的所有队列上。

* 很像子网广播，每台子网内的主机都获得了一份复制的消息。Fanout交换机转发消息是最快的。

* Topic Exchange – 将路由键和某模式进行匹配。此时队列需要绑定要一个模式上。符号“#”匹配一个或多个词，符号“*”匹配不多不少一个词。

* 因此“audit.#”能够匹配到“audit.irs.corporate”，但是“audit.*” 只会匹配到“audit.irs”。

*/

channel.exchangeDeclare(EXCHANGE_NAME, "fanout");

//创建一个非持久的、唯一的且自动删除的队列且队列名称由服务器随机产生一般情况这个名称与amq.gen-JzTY20BRgKO-HjmUJj0wLg 类似。

String queueName = channel.queueDeclare().getQueue();

// 为转发器指定队列，设置binding

channel.queueBind(queueName, EXCHANGE_NAME, "");

System.out.println(" [*] Waiting for messages. To exit press CTRL+C");

QueueingConsumer consumer = new QueueingConsumer(channel);

// 指定接收者，第二个参数为自动应答，无需手动应答

channel.basicConsume(queueName, true, consumer);

while (true) {

QueueingConsumer.Delivery delivery = consumer.nextDelivery();

String message = new String(delivery.getBody());

print2File(message);

}

}

private static void print2File(String msg) {

try {

String dir = ReceiveLogsToFile.class.getClassLoader().getResource("").getPath();

String logFileName = new SimpleDateFormat("yyyy-MM-dd").format(new Date());

File file = new File(dir, logFileName + ".txt");

FileOutputStream fos = new FileOutputStream(file, true);

fos.write((msg + "\r\n").getBytes());

fos.flush();

fos.close();

} catch (FileNotFoundException e) {

e.printStackTrace();

} catch (IOException e) {

e.printStackTrace();

}

}

}

3.接受写控制台代码ReceiveLogsToConsole.java

创建一个队列，然后将队列与转发器绑定，然后将消费者与该队列绑定，然后打印到控制台。

[java] view plain copy

print?

package cn.slimsmart.rabbitmq.demo.fanout;

import com.rabbitmq.client.AMQP;

import com.rabbitmq.client.Channel;

import com.rabbitmq.client.Connection;

import com.rabbitmq.client.ConnectionFactory;

import com.rabbitmq.client.QueueingConsumer;

public class ReceiveLogsToConsole {

private final static String EXCHANGE_NAME = "ex_log";

public static void main(String[] args) throws Exception {

// 创建连接和频道

ConnectionFactory factory = new ConnectionFactory();

factory.setHost("192.168.101.174");

// 指定用户 密码

factory.setUsername("admin");

factory.setPassword("admin");

// 指定端口

factory.setPort(AMQP.PROTOCOL.PORT);

Connection connection = factory.newConnection();

Channel channel = connection.createChannel();

channel.exchangeDeclare(EXCHANGE_NAME, "fanout");

// 创建一个非持久的、唯一的且自动删除的队列且队列名称由服务器随机产生一般情况这个名称与amq.gen-JzTY20BRgKO-HjmUJj0wLg 类似。

String queueName = channel.queueDeclare().getQueue();

// 为转发器指定队列，设置binding

channel.queueBind(queueName, EXCHANGE_NAME, "");

System.out.println(" [*] Waiting for messages. To exit press CTRL+C");

QueueingConsumer consumer = new QueueingConsumer(channel);

// 指定接收者，第二个参数为自动应答，无需手动应答

channel.basicConsume(queueName, true, consumer);

while (true)

{

QueueingConsumer.Delivery delivery = consumer.nextDelivery();

String message = new String(delivery.getBody());

System.out.println(" [x] Received '" + message + "'");

}

}

}

先启动2个消费者，再运行生产者观察消费者运行情况，可见2个消费者都受到相同的消息。生产者将消息发送至转发器，转发器决定将消息发送至哪些队列，消费者绑定队列获取消息。