[rabbitmq] 安装与配置 javaexchangeerlang
程序员文章站
2022-05-30 16:17:33
...
RabbitMQ
-------------------------
1
下载
erlang
http://www.erlang.org/download/otp_win32_17.3.exe
http://www.erlang.org/download/otp_src_17.3.tar.gz
http://www.erlang.org/download/otp_doc_html_17.3.tar.gz
http://www.erlang.org/download/otp_doc_man_17.3.tar.gz
rabbitmq server
http://www.rabbitmq.com/releases/rabbitmq-server/v3.4.2/rabbitmq-server-3.4.2.exe
rabbitmq java client
http://www.rabbitmq.com/releases/rabbitmq-java-client/v3.4.2/rabbitmq-java-client-bin-3.4.2.zip
http://www.rabbitmq.com/releases/rabbitmq-java-client/v3.4.2/rabbitmq-java-client-3.4.2.zip
2
配置
3
基本概念
http://www.rabbitmq.com/tutorials/tutorial-one-java.html
RabbitMQ 是一个消息代理,它接收生产者的消息,然后转发给消费者.同时,它路由,缓存,持久这个消息根据设定的规则.
producer 生产者,只负责发送消息.消息发送到指定队列中.
queue 队列或邮箱,它存储消息,它的大小没有限制,只要磁盘空间足够.消息只能缓存在队列中,队列属于消息代理的一部分.消息代理还包括 exchange 等.
consumer 消费者,它从队列接收消息.
AMQP:高级消息队列协议.
P(producer) -- msg --> |Queue|Queue|Queue| -- msg --> C(consumer)
PS:
生产者,消息代理,消费者通常不在同一机器上.
4 Hello World
----------------
我们这里使用 Java 客户端.
STEP 1:
P(producer) -- msg --> |HelloQueue|
STEP 2
|HelloQueue| -- msg --> C(consumer)
5
Work Queue
P -- MSG -- |Q| -- MSG -- Worker1|Worker2
Work1 , Work2 轮流获取消息处理,
为避免Worker1消息单节点处理失败, 修改ACK返回时机为消息处理成功后.ACK返回,消息代理才会删除已处理的消息,否则是Worker1一接收到就删除.
为保证消息代理节点失败,导致消息丢失, 设置消息持久化参数:
6
发布/订阅模式
一个消息多个接收者接收.
EXCHANGE: 消息交换,用于路由/过滤消息.
P -- MSG -- |X| -- |Q1| -- MSG -- C1
-- |Q2| -- MSG -- C2
7
路由
绑定: 一个队列对交换的消息感兴趣.
绑定KEY: 绑定时设置
路由KEY: 发布时设置 routingKey, fanout 会简单忽略这个值.
交换策略:
direct exchange: 直接交换,根据 routingKey 来选择发布到哪些队列.比fanout更加灵活.
fanout exchange: 扇出交换(全部队列都会接收这个消息)
topic exchange: routingKey用 "." 号分开.根据 routingKey 来选择发布到哪些匹配的队列, 支持模糊匹配.
8
RPC 远程过程调用
通常发送方要等待接收方的处理结果.
发送方阻塞直到方法返回结, 每个客户端一个响应队列.
为了识别哪个响应属于哪个请求,要加个请求ID.
result = Client.call(request); // block until result has returned
Client -- id:RequestId , replyTo:ReplyQueue -- HelloQueue -- Server
| |
-- id:RequestId , replyTo:ReplyQueue -- ReplyQueue ------
RPCServer
RPCClient
-------------------------
1
下载
erlang
http://www.erlang.org/download/otp_win32_17.3.exe
http://www.erlang.org/download/otp_src_17.3.tar.gz
http://www.erlang.org/download/otp_doc_html_17.3.tar.gz
http://www.erlang.org/download/otp_doc_man_17.3.tar.gz
rabbitmq server
http://www.rabbitmq.com/releases/rabbitmq-server/v3.4.2/rabbitmq-server-3.4.2.exe
rabbitmq java client
http://www.rabbitmq.com/releases/rabbitmq-java-client/v3.4.2/rabbitmq-java-client-bin-3.4.2.zip
http://www.rabbitmq.com/releases/rabbitmq-java-client/v3.4.2/rabbitmq-java-client-3.4.2.zip
2
配置
3
基本概念
http://www.rabbitmq.com/tutorials/tutorial-one-java.html
RabbitMQ 是一个消息代理,它接收生产者的消息,然后转发给消费者.同时,它路由,缓存,持久这个消息根据设定的规则.
producer 生产者,只负责发送消息.消息发送到指定队列中.
queue 队列或邮箱,它存储消息,它的大小没有限制,只要磁盘空间足够.消息只能缓存在队列中,队列属于消息代理的一部分.消息代理还包括 exchange 等.
consumer 消费者,它从队列接收消息.
AMQP:高级消息队列协议.
P(producer) -- msg --> |Queue|Queue|Queue| -- msg --> C(consumer)
PS:
生产者,消息代理,消费者通常不在同一机器上.
4 Hello World
----------------
我们这里使用 Java 客户端.
STEP 1:
P(producer) -- msg --> |HelloQueue|
public class P { private final static String QUEUE_NAME = "HelloQueue"; public static void main(String[] args) throws Exception { // 建立连接 ConnectionFactory factory = new ConnectionFactory(); factory.setHost("127.0.0.1"); Connection connection = factory.newConnection(); Channel channel = connection.createChannel(); // 声明队列 channel.queueDeclare(QUEUE_NAME, false, false, false, null); // 创建消息 String message = "Hello World!"; // 发布消息到 HelloQueue channel.basicPublish("", QUEUE_NAME, null, message.getBytes()); // 关闭连接 channel.close(); connection.close(); } }
STEP 2
|HelloQueue| -- msg --> C(consumer)
public class C { private final static String QUEUE_NAME = "HelloQueue"; public static void main(String[] args) throws Exception { // 建立连接 ConnectionFactory factory = new ConnectionFactory(); factory.setHost("localhost"); Connection connection = factory.newConnection(); Channel channel = connection.createChannel(); // 声明队列 channel.queueDeclare(QUEUE_NAME, false, false, false, null); // 创建消费者,处理消息 QueueingConsumer consumer = new QueueingConsumer(channel); channel.basicConsume(QUEUE_NAME, true, consumer); while (true) { // block method ? yes. // 阻塞直到接收到下一条消息 QueueingConsumer.Delivery delivery = consumer.nextDelivery(); String message = new String(delivery.getBody()); System.out.println("Received : " + message); } }
5
Work Queue
P -- MSG -- |Q| -- MSG -- Worker1|Worker2
Work1 , Work2 轮流获取消息处理,
为避免Worker1消息单节点处理失败, 修改ACK返回时机为消息处理成功后.ACK返回,消息代理才会删除已处理的消息,否则是Worker1一接收到就删除.
boolean autoAck = false; channel.basicConsume("hello", autoAck, consumer); while (true) { QueueingConsumer.Delivery delivery = consumer.nextDelivery(); //... channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false); }
为保证消息代理节点失败,导致消息丢失, 设置消息持久化参数:
boolean durable = true; channel.queueDeclare("hello", durable, false, false, null); ... channel.basicPublish( "", TASK_QUEUE_NAME, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes());
6
发布/订阅模式
一个消息多个接收者接收.
EXCHANGE: 消息交换,用于路由/过滤消息.
P -- MSG -- |X| -- |Q1| -- MSG -- C1
-- |Q2| -- MSG -- C2
// 交换策略扇出(fanout):消息交换到所有队列 channel.exchangeDeclare(EXCHANGE_NAME, "fanout"); String message = getMessage(argv); channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes()); ...
channel.exchangeDeclare(EXCHANGE_NAME, "fanout"); // 临时队列.随机命名的队列 String queueName = channel.queueDeclare().getQueue(); // 绑定交换与队列 channel.queueBind(queueName, EXCHANGE_NAME, "");
7
路由
绑定: 一个队列对交换的消息感兴趣.
绑定KEY: 绑定时设置
路由KEY: 发布时设置 routingKey, fanout 会简单忽略这个值.
交换策略:
direct exchange: 直接交换,根据 routingKey 来选择发布到哪些队列.比fanout更加灵活.
fanout exchange: 扇出交换(全部队列都会接收这个消息)
topic exchange: routingKey用 "." 号分开.根据 routingKey 来选择发布到哪些匹配的队列, 支持模糊匹配.
channel.exchangeDeclare(EXCHANGE_NAME, "topic"); ... channel.basicPublish(EXCHANGE_NAME, routingKey, null, message.getBytes());
channel.exchangeDeclare(EXCHANGE_NAME, "topic"); String queueName = channel.queueDeclare().getQueue(); for(String bindingKey : bindingKeys){ channel.queueBind(queueName, EXCHANGE_NAME, bindingKey); } QueueingConsumer consumer = new QueueingConsumer(channel); channel.basicConsume(queueName, true, consumer); while (true) { QueueingConsumer.Delivery delivery = consumer.nextDelivery(); String message = new String(delivery.getBody()); String routingKey = delivery.getEnvelope().getRoutingKey(); ... }
8
RPC 远程过程调用
通常发送方要等待接收方的处理结果.
发送方阻塞直到方法返回结, 每个客户端一个响应队列.
为了识别哪个响应属于哪个请求,要加个请求ID.
result = Client.call(request); // block until result has returned
Client -- id:RequestId , replyTo:ReplyQueue -- HelloQueue -- Server
| |
-- id:RequestId , replyTo:ReplyQueue -- ReplyQueue ------
RPCServer
private static final String RPC_QUEUE_NAME = "rpc_queue"; ConnectionFactory factory = new ConnectionFactory(); factory.setHost("localhost"); Connection connection = factory.newConnection(); Channel channel = connection.createChannel(); channel.queueDeclare(RPC_QUEUE_NAME, false, false, false, null); channel.basicQos(1); QueueingConsumer consumer = new QueueingConsumer(channel); channel.basicConsume(RPC_QUEUE_NAME, false, consumer); System.out.println(" [x] Awaiting RPC requests"); while (true) { QueueingConsumer.Delivery delivery = consumer.nextDelivery(); BasicProperties props = delivery.getProperties(); BasicProperties replyProps = new BasicProperties .Builder() .correlationId(props.getCorrelationId()) .build(); String message = new String(delivery.getBody()); int n = Integer.parseInt(message); System.out.println(" [.] fib(" + message + ")"); String response = "" + fib(n); channel.basicPublish( "", props.getReplyTo(), replyProps, response.getBytes()); channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false); }
RPCClient
private Connection connection; private Channel channel; private String requestQueueName = "rpc_queue"; private String replyQueueName; private QueueingConsumer consumer; public RPCClient() throws Exception { ConnectionFactory factory = new ConnectionFactory(); factory.setHost("localhost"); connection = factory.newConnection(); channel = connection.createChannel(); replyQueueName = channel.queueDeclare().getQueue(); consumer = new QueueingConsumer(channel); channel.basicConsume(replyQueueName, true, consumer); } public String call(String message) throws Exception { String response = null; String corrId = java.util.UUID.randomUUID().toString(); BasicProperties props = new BasicProperties .Builder() .correlationId(corrId) .replyTo(replyQueueName) .build(); channel.basicPublish("", requestQueueName, props, message.getBytes()); while (true) { QueueingConsumer.Delivery delivery = consumer.nextDelivery(); if (delivery.getProperties().getCorrelationId().equals(corrId)) { response = new String(delivery.getBody()); break; } } return response; } public void close() throws Exception { connection.close(); }
上一篇: 注册、登陆、退出请求分发器实例
下一篇: css样式和选择器的使用方法