RabbitMQ In JAVA 介绍及使用

介绍:

  RabbitMQ是开源的消息中间件,它是轻量级的,支持多种消息传递协议,可以部署在分布式和联合配置中,以满足高级别、高可用性需求。并且可在许多操作系统和云环境上运行,并为大多数流行语言提供了广泛的开发工具。(这里只介绍JAVA下的RabbitMQ的使用,感兴趣的可以查看官方文档:http://www.rabbitmq.com/getstarted.html);

安装:

  参考:http://www.cnblogs.com/lfalex0831/p/8951955.html(windows安装)

应用场景:

  1、异步处理,主要为了较少请求的响应时间和解耦。即将比较耗时又不需要同步的操作放入消息队列中进行传递请求,只要保证消息格式(可以理解为接头的暗号)不变,这样消息的发送方和接收方互不干扰交互,即为解耦;

  2、广播,顾名思义,广播的好处就是一次发送,大家共享,大大的减少了冗余的操作,也降低了新增消费者的成本;

  3、流量削峰,比如秒杀活动,因为流量过大,导致应用挂掉,为了避免这个问题,会在应用前端加入消息队列。 
  作用: 
    1.可以控制进入后台的服务,超过阀值的订单直接丢弃;
    2.可以缓解瞬时的高流量压垮应用;

  ps:秒杀系统优化思路可以从将请求尽量拦截在系统上游充分利用缓存;

  (如果还有别的应用场景,请大家多多指教。。。)

使用场景(本文使用的RabbitMQ的版本为5.20版本):

This tutorial assumes RabbitMQ is installed and running on localhost on standard port (). In case you use a different host, port or credentials, connections settings would require adjusting.

注意:根据官方文档说明,RabbitMQ的默认访问端口为5672,而管理端口为15672,希望不要搞混了(我刚接触时就没注意,果断乱了。。-_-||)。

  基本概念:

  • Broker:简单来说就是消息队列服务器实体
  • Exchange:消息交换机,它指定消息按什么规则,路由到哪个队列
  • Queue:消息队列载体,每个消息都会被投入到一个或多个队列
  • Binding:绑定,它的作用就是把exchange和queue按照路由规则绑定起来
  • Routing Key:路由关键字,exchange根据这个关键字进行消息投递
  • vhost:虚拟主机,一个broker里可以开设多个vhost,用作不同用户的权限分离
  • producer:消息生产者,就是投递消息的程序
  • consumer:消息消费者,就是接受消息的程序
  • channel:消息通道,在客户端的每个连接里,可建立多个channel,每个channel代表一个会话任务

  1、简单使用(HelloWorld)

  在下图中,P是我们的生产者,C是我们的消费者。中间的框是一个队列——RabbitMQ代表消费者保存的消息缓冲区。

   RabbitMQ In JAVA 介绍及使用

  创建RabbitMQ的工厂类:

 import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import java.io.IOException;
import java.util.concurrent.TimeoutException; public class ConnectionUtil { private static final String RABBIT_HOST = "localhost"; private static final String RABBIT_USERNAME = "guest"; private static final String RABBIT_PASSWORD = "guest"; private static Connection connection = null; public static Connection getConnection() {
if(connection == null) {
ConnectionFactory connectionFactory = new ConnectionFactory();
connectionFactory.setHost(RABBIT_HOST);
connectionFactory.setUsername(RABBIT_USERNAME);
connectionFactory.setPassword(RABBIT_PASSWORD);
try {
connection = connectionFactory.newConnection();
} catch (IOException e) {
e.printStackTrace();
} catch (TimeoutException e) {
e.printStackTrace();
}
}
return connection;
} }

ConnectionUtil.java

  创建生产者Producer:

 import com.cn.ConnectionUtil;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import java.io.IOException;
import java.util.concurrent.TimeoutException; public class Producer { private static final String QUEUE_NAME="test_queue"; public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
//获取连接
Connection connection = ConnectionUtil.getConnection();
System.out.println(connection);
//创建通道
Channel channel = connection.createChannel(1);
/*
* 声明(创建)队列
* 参数1:队列名称
* 参数2:为true时server重启队列不会消失
* 参数3:队列是否是独占的,如果为true只能被一个connection使用,其他连接建立时会抛出异常
* 参数4:队列不再使用时是否自动删除(没有连接,并且没有未处理的消息)
* 参数5:建立队列时的其他参数
*/
channel.queueDeclare(QUEUE_NAME,false,false,false,null);
String message = "Hello World!";
for (int i = 0; i < 20; i++) {
message = message + i;
channel.basicPublish("",QUEUE_NAME,null,message.getBytes());
Thread.sleep(1000);
}
System.out.println("生产者 send :"+message);
channel.close();
connection.close();
} }

Producer.java

  创建消费者Consumer:

 import com.cn.ConnectionUtil;
import com.rabbitmq.client.AMQP.BasicProperties;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;
import java.io.IOException; public class Consumer { private static final String QUEUE_NAME = "test_queue"; public static void main(String[] args) throws IOException {
Connection connection = ConnectionUtil.getConnection();
Channel channel = connection.createChannel(1);
channel.queueDeclare(QUEUE_NAME,false,false,false,null);
StringBuffer message = new StringBuffer();
//自4.0+ 版本后无法再使用QueueingConsumer,而官方推荐使用DefaultConsumer
com.rabbitmq.client.Consumer consumer = new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body)
throws IOException {
try {
Thread.sleep(2000);
} catch (InterruptedException e) {
e.printStackTrace();
}
super.handleDelivery(consumerTag, envelope, properties, body);
message.append(new String(body,"UTF-8"));
System.out.println(new String(body,"UTF-8"));
}
};
//监听队列,当b为true时,为自动提交(只要消息从队列中获取,无论消费者获取到消息后是否成功消息,都认为是消息已经成功消费),
// 当b为false时,为手动提交(消费者从队列中获取消息后,服务器会将该消息标记为不可用状态,等待消费者的反馈,
// 如果消费者一直没有反馈,那么该消息将一直处于不可用状态。
//如果选用自动确认,在消费者拿走消息执行过程中出现宕机时,消息可能就会丢失!!)
//使用channel.basicAck(envelope.getDeliveryTag(),false);进行消息确认
channel.basicConsume(QUEUE_NAME,true,consumer);
System.out.println(message.toString());
}
}

Consumer.java

  测试结果,Consumer收到Producer的消息。

  2、Work Queue

  上一个例子是一对一发送接收形式,而工作队列为一对多发送接收形式。工作队列(即任务队列)背后的主要思想是避免立即执行资源密集型任务,并且必须等待它完成。相反,我们把任务安排在以后做。我们将任务封装为消息并将其发送到队列。在后台运行的工作进程会弹出任务并最终执行任务。当你运行许多Consumer时,任务将在他们之间共享,如下图:

   RabbitMQ In JAVA 介绍及使用

  由于工厂类已经创建,直接使用即可。

  创建生产者Producer:

package com.cn.work;

import com.cn.ConnectionUtil;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection; /**
* @program: rabbit-learn
* @description: 生产者
* @create: 2018-04-26 16:18
**/
public class Producer { private final static String QUEUE_NAME = "test_queue_work"; public static void main(String[] args) throws Exception {
Connection connection = ConnectionUtil.getConnection();
Channel channel = connection.createChannel();
channel.queueDeclare(QUEUE_NAME,false,false,false,null);
for (int i = 0; i < 50; i++) {
String message = "" + i;
channel.basicPublish("",QUEUE_NAME,null,message.getBytes());
Thread.sleep(100 * i);
}
channel.close();
connection.close();
} }

Producer.java

  创建消费者1,2:

import com.cn.ConnectionUtil;
import com.rabbitmq.client.AMQP.BasicProperties;
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 java.io.IOException; public class Consumer1 { private final static String QUEUE_NAME = "test_queue_work"; public static void main(String[] args) throws IOException {
Connection connection = ConnectionUtil.getConnection();
Channel channel = connection.createChannel();
channel.basicQos(1);//能者多劳模式
//声明队列
channel.queueDeclare(QUEUE_NAME,false,false,false,null); //自4.0+ 版本后无法再使用QueueingConsumer,而官方推荐使用DefaultConsumer
Consumer consumer = new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body)
throws IOException {
super.handleDelivery(consumerTag, envelope, properties, body);
String message = new String(body,"UTF-8");
System.out.println(message);
try {
Thread.sleep(10);
} catch (InterruptedException e) {
e.printStackTrace();
}
try {
doWork(message);
}finally {
channel.basicAck(envelope.getDeliveryTag(),false);
}
}
};
//监听队列,当b为true时,为自动提交(只要消息从队列中获取,无论消费者获取到消息后是否成功消息,都认为是消息已经成功消费),
// 当b为false时,为手动提交(消费者从队列中获取消息后,服务器会将该消息标记为不可用状态,等待消费者的反馈,
// 如果消费者一直没有反馈,那么该消息将一直处于不可用状态。
//如果选用自动确认,在消费者拿走消息执行过程中出现宕机时,消息可能就会丢失!!)
//使用channel.basicAck(envelope.getDeliveryTag(),false);进行消息确认
channel.basicConsume(QUEUE_NAME,false,consumer);
} private static void doWork(String task) {
for (char ch : task.toCharArray()) {
if (ch == '.') {
try {
Thread.sleep(1000);
} catch (InterruptedException _ignored) {
Thread.currentThread().interrupt();
}
}
}
} }

Consumer1.java

 import com.cn.ConnectionUtil;
import com.rabbitmq.client.AMQP.BasicProperties;
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 java.io.IOException; public class Consumer2 { private final static String QUEUE_NAME = "test_queue_work"; public static void main(String[] args) throws IOException {
Connection connection = ConnectionUtil.getConnection();
Channel channel = connection.createChannel();
channel.basicQos(1);//能者多劳模式
//声明队列
channel.queueDeclare(QUEUE_NAME,false,false,false,null);
//自4.0+ 版本后无法再使用QueueingConsumer,而官方推荐使用DefaultConsumer
Consumer consumer = new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body)
throws IOException {
super.handleDelivery(consumerTag, envelope, properties, body);
String message = new String(body,"UTF-8");
System.out.println(message);
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
try {
doWork(message);
}finally {
channel.basicAck(envelope.getDeliveryTag(), false);
}
}
};
//监听队列,当b为true时,为自动提交(只要消息从队列中获取,无论消费者获取到消息后是否成功消息,都认为是消息已经成功消费),
// 当b为false时,为手动提交(消费者从队列中获取消息后,服务器会将该消息标记为不可用状态,等待消费者的反馈,
// 如果消费者一直没有反馈,那么该消息将一直处于不可用状态。
//如果选用自动确认,在消费者拿走消息执行过程中出现宕机时,消息可能就会丢失!!)
//使用channel.basicAck(envelope.getDeliveryTag(),false);进行消息确认
channel.basicConsume(QUEUE_NAME,false,consumer);
} /**
* @Description: 业务代码
* @Param:
* @return:
* @Author: 535504
* @Date: 2018/4/26
*/
private static void doWork(String task) {
for (char ch : task.toCharArray()) {
if (ch == '.') {
try {
Thread.sleep(1000);
} catch (InterruptedException _ignored) {
Thread.currentThread().interrupt();
}
}
}
} }

Consumer2.java

  测试结果,当消费者中的channel.basicQos(1);这行代码的注释打开时,执行会发现,休眠时间短的消费者执行的任务多,而注释后,再次执行会发现消费者1和消费者2获取到的消息内容是不同的,同一个消息只能被一个消费者获取,消费者1和消费者2获取到的消息的数量是相同的,一个是奇数一个是偶数。

  3、Publish/Subscribe(严格来说下面介绍的路由和通配符模式也是发布订阅)

  在发布订阅模式中,消息需要发送到MQ的交换机exchange上,exchange根据配置的路由方式发到相应的Queue上,Queue又将消息发送给consumer,消息从queue到consumer, 消息队列的使用过程大概如下:
  1.客户端连接到消息队列服务器,打开一个channel。
  2.客户端声明一个exchange,并设置相关属性。
  3.客户端声明一个queue,并设置相关属性。
  4.客户端在exchange和queue之间建立好绑定关系。
  5.客户端投递消息到exchange。

  RabbitMQ In JAVA 介绍及使用

  创建生产者Producer: 

 import com.cn.ConnectionUtil;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import java.io.IOException;
import java.util.concurrent.TimeoutException; /**
* @program: rabbit-learn
* @description: 生产者,订阅模式
* @author:
* @create:
* 消息发送到没有队列绑定的交换机时,消息将丢失,因为,交换机没有存储消息的能力,消息只能存在在队列中。
**/
public class Producer { //交换机名称
private static final String EXCHANGE_NAME = "test_exchange_fanout"; public static void main(String[] args) throws IOException, TimeoutException {
Connection connection = ConnectionUtil.getConnection();
Channel channel = connection.createChannel();
/*
声明exchange交换机
参数1:交换机名称
参数2:交换机类型
参数3:交换机持久性,如果为true则服务器重启时不会丢失
参数4:交换机在不被使用时是否删除
参数5:交换机的其他属性
*/
channel.exchangeDeclare(EXCHANGE_NAME,"fanout", true,true,null); String message = "订阅消息";
channel.basicPublish(EXCHANGE_NAME,"",null,message.getBytes());
System.out.println("生产者 send :"+message);
channel.close();
connection.close();
} }

Producer.java

  创建消费者Consumer1、2:

 import com.cn.ConnectionUtil;
import com.rabbitmq.client.AMQP.BasicProperties;
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 java.io.IOException; public class Consumer1 { private static final String QUEUE_NAME = "test_queue_exchange_1"; private static final String EXCHANGE_NAME = "test_exchange_fanout"; public static void main(String[] args) throws IOException {
Connection connection = ConnectionUtil.getConnection();
Channel channel = connection.createChannel(); //声明队列
channel.queueDeclare(QUEUE_NAME,false,false,false,null); /*
绑定队列到交换机(这个交换机名称一定要和生产者的交换机名相同)
参数1:队列名
参数2:交换机名
参数3:Routing key 路由键
*/
channel.queueBind(QUEUE_NAME,EXCHANGE_NAME,""); //同一时刻服务器只会发一条数据给消费者
channel.basicQos(1); Consumer consumer = new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body)
throws IOException {
super.handleDelivery(consumerTag, envelope, properties, body);
String message = new String(body,"UTF-8");
System.out.println("收到消息:"+message);
channel.basicAck(envelope.getDeliveryTag(),false);
}
};
channel.basicConsume(QUEUE_NAME,false,consumer);
} }

Consumer1.java

 import com.cn.ConnectionUtil;
import com.rabbitmq.client.AMQP.BasicProperties;
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 java.io.IOException; public class Consumer2 { private static final String QUEUE_NAME = "test_queue_exchange_2"; private static final String EXCHANGE_NAME = "test_exchange_fanout"; public static void main(String[] args) throws IOException {
Connection connection = ConnectionUtil.getConnection();
Channel channel = connection.createChannel(); //声明队列
channel.queueDeclare(QUEUE_NAME,false,false,false,null); /*
绑定队列到交换机(这个交换机名称一定要和生产者的交换机名相同)
参数1:队列名
参数2:交换机名
参数3:Routing key 路由键
*/
channel.queueBind(QUEUE_NAME,EXCHANGE_NAME,""); //同一时刻服务器只会发一条数据给消费者
channel.basicQos(1); Consumer consumer = new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body)
throws IOException {
super.handleDelivery(consumerTag, envelope, properties, body);
String message = new String(body,"UTF-8");
System.out.println("收到消息:"+message);
channel.basicAck(envelope.getDeliveryTag(),false);
}
};
channel.basicConsume(QUEUE_NAME,false,consumer);
} }

Consumer2.java

  4、Routing(路由)

  根据指定的路由键发送到对应的消息队列中,如下图,在这个设置中,我们可以看到与它绑定的两个队列的直接交换X。第一个队列绑定了绑定键橙色,第二个队列有两个绑定,一个绑定键为黑色,另一个为绿色。在这样的设置中,将发送到与路由键橙色的交换的消息将被路由到队列Q1。带有黑色或绿色的路由键的消息将会进入Q2。所有其他消息将被丢弃。

    RabbitMQ In JAVA 介绍及使用

  创建生产者Producer:

 import com.cn.ConnectionUtil;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection; public class Producer { private static final String EXCHANGE_NAME = "test_exchange_direct"; public static void main(String[] argv) throws Exception {
// 获取到连接以及mq通道
Connection connection = ConnectionUtil.getConnection();
Channel channel = connection.createChannel(); // 声明exchange,路由模式声明direct
channel.exchangeDeclare(EXCHANGE_NAME, "direct"); // 消息内容
String message = "这是消息B";
channel.basicPublish(EXCHANGE_NAME, "B", null, message.getBytes());
String messageA = "这是消息A";
channel.basicPublish(EXCHANGE_NAME, "A", null, messageA.getBytes());
System.out.println(" [生产者] Sent '" + message + "'"); channel.close();
connection.close();
} }

Producer.java

  创建消费者Consumer1、2:

 package com.cn.routing;

 import com.cn.ConnectionUtil;
import com.rabbitmq.client.AMQP.BasicProperties;
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 java.io.IOException; /**
* @program: rabbit-learn
* @description: 消费者1
* @author: 535504
* @create: 2018-04-26 17:52
**/
public class Consumer1 {
private final static String QUEUE_NAME = "test_queue_direct_A"; private final static String EXCHANGE_NAME = "test_exchange_direct"; public static void main(String[] argv) throws Exception { // 获取到连接以及mq通道
Connection connection = ConnectionUtil.getConnection();
Channel channel = connection.createChannel(); // 声明队列
//channel.queueDeclare(QUEUE_NAME, false, false, false, null);
channel.exchangeDeclare(EXCHANGE_NAME,"direct");
/*
* 绑定队列到交换机
* 参数1:队列的名称
* 参数2:交换机的名称
* 参数3:routingKey
*/
channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "A"); // 同一时刻服务器只会发一条消息给消费者
channel.basicQos(1); // 定义队列的消费者
Consumer consumer = new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body)
throws IOException {
super.handleDelivery(consumerTag, envelope, properties, body);
System.out.println(new String(body,"UTF-8"));
}
};
channel.basicConsume(QUEUE_NAME,true,consumer);
}
}

Consumer1.java

 import com.cn.ConnectionUtil;
import com.rabbitmq.client.AMQP.BasicProperties;
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 java.io.IOException; public class Consumer2 {
private final static String QUEUE_NAME = "test_queue_direct_B"; private final static String EXCHANGE_NAME = "test_exchange_direct"; public static void main(String[] argv) throws Exception { // 获取到连接以及mq通道
Connection connection = ConnectionUtil.getConnection();
Channel channel = connection.createChannel(); // 声明队列
//channel.queueDeclare(QUEUE_NAME, false, false, false, null);
channel.exchangeDeclare(EXCHANGE_NAME,"direct");
/*
* 绑定队列到交换机
* 参数1:队列的名称
* 参数2:交换机的名称
* 参数3:routingKey
*/
channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "B"); // 同一时刻服务器只会发一条消息给消费者
channel.basicQos(1); // 定义队列的消费者
Consumer consumer = new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body)
throws IOException {
super.handleDelivery(consumerTag, envelope, properties, body);
System.out.println(new String(body,"UTF-8"));
}
};
channel.basicConsume(QUEUE_NAME,true,consumer);
}
}

Consumer2.java

  5、Topics(主题通配符)

  可以理解为Routing的通配符模式,如下图:

  RabbitMQ In JAVA 介绍及使用

  “#”:表示匹配一个或多个词;(lazy.a.b.c)

  “*”:表示匹配一个词;(a.orange.b)

  创建生产者Producer:

 import com.cn.ConnectionUtil;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection; import java.io.IOException;
import java.util.concurrent.TimeoutException; public class Producer { private static final String EXCHANGE_NAME = "test_exchange_topic"; public static void main(String[] args) throws IOException, TimeoutException {
Connection connection = ConnectionUtil.getConnection();
Channel channel = connection.createChannel();
//声明交换机
channel.exchangeDeclare(EXCHANGE_NAME,"topic");
String message = "匹配insert";
channel.basicPublish(EXCHANGE_NAME,"order.update",false,false,null,message.getBytes()); channel.close();
connection.close();
}
}

Producer.java

  创建消费者Consumer1、2:

 import com.cn.ConnectionUtil;
import com.rabbitmq.client.AMQP.BasicProperties;
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 java.io.IOException; public class Consumer1 { private static final String EXCHANGE_NAME = "test_exchange_topic"; private static final String QUEUE_NAME = "test_queue_topic_1"; public static void main(String[] args) throws IOException {
Connection connection = ConnectionUtil.getConnection();
Channel channel = connection.createChannel();
//channel.queueDeclare(QUEUE_NAME,false,false,false,null);
channel.exchangeDeclare(EXCHANGE_NAME,"topic");
//order.#
channel.queueBind(QUEUE_NAME,EXCHANGE_NAME,"order.*"); channel.basicQos(1); Consumer consumer = new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body)
throws IOException {
super.handleDelivery(consumerTag, envelope, properties, body);
System.out.println(new String(body,"UTF-8"));
}
};
channel.basicConsume(QUEUE_NAME,true,consumer); }
}

Consumer1.java

 import com.cn.ConnectionUtil;
import com.rabbitmq.client.AMQP.BasicProperties;
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 java.io.IOException; public class Consumer2 { private static final String EXCHANGE_NAME = "test_exchange_topic"; private static final String QUEUE_NAME = "test_queue_topic_2"; public static void main(String[] args) throws IOException {
Connection connection = ConnectionUtil.getConnection();
Channel channel = connection.createChannel();
channel.queueDeclare(QUEUE_NAME,false,false,false,null); channel.queueBind(QUEUE_NAME,EXCHANGE_NAME,"order.insert"); channel.basicQos(1); Consumer consumer = new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body) throws IOException {
super.handleDelivery(consumerTag, envelope, properties, body);
System.out.println("接收消息:" + new String(body, "UTF-8"));
}
};
channel.basicConsume(QUEUE_NAME,true,consumer); }
}

Consumer2.java

  6、RPC(远程调用)

  如果我们需要在远程计算机上运行一个函数并等待结果,这种模式通常称为远程过程调用或RPC;

   RabbitMQ In JAVA 介绍及使用

  创建RPC服务:

 import com.cn.ConnectionUtil;
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 java.io.IOException;
import java.util.concurrent.TimeoutException; public class RPCServer { private static final String RPC_QUEUE_NAME = "rpc_queue"; public static void main(String[] args) throws IOException, TimeoutException {
Connection connection = ConnectionUtil.getConnection();
final Channel channel = connection.createChannel();
channel.queueDeclare(RPC_QUEUE_NAME, false, false, false, null);
channel.basicQos(1); Consumer consumer = new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
super.handleDelivery(consumerTag, envelope, properties, body);
AMQP.BasicProperties properties1 = new AMQP.BasicProperties.Builder().correlationId(properties.getCorrelationId()).build();
String mes = new String(body, "UTF-8");
int num = Integer.valueOf(mes);
System.out.println("接收数据:" + num);
num = fib(num);
channel.basicPublish("", properties.getReplyTo(), properties1, String.valueOf(num).getBytes());
channel.basicAck(envelope.getDeliveryTag(), false);
}
};
channel.basicConsume(RPC_QUEUE_NAME, false, consumer);
while (true) {
synchronized (consumer) {
try {
consumer.wait();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
} /*
获取斐波那契数列的第n个值得大小
*/
private static int fib(int n) {
System.out.println(n);
if (n == 0)
return 0;
if (n == 1)
return 1;
return fib(n - 1) + fib(n - 2);
}
}

RPCServer.java

  创建RPC客户端:

 import com.cn.ConnectionUtil;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope; import java.io.IOException;
import java.util.UUID;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeoutException; public class RPCClient { private Connection connection;
private Channel channel;
private String requestQueueName = "rpc_queue";
private String replyQueueName; public RPCClient() throws IOException, TimeoutException {
connection = ConnectionUtil.getConnection();
channel = connection.createChannel(); replyQueueName = channel.queueDeclare().getQueue();
} public String call(String message) throws IOException, InterruptedException {
final String corrId = UUID.randomUUID().toString(); AMQP.BasicProperties props = new AMQP.BasicProperties
.Builder()
.correlationId(corrId)
.replyTo(replyQueueName)
.build(); channel.basicPublish("", requestQueueName, props, message.getBytes("UTF-8")); final BlockingQueue<String> response = new ArrayBlockingQueue<String>(1); channel.basicConsume(replyQueueName, true, new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
if (properties.getCorrelationId().equals(corrId)) {
response.offer(new String(body, "UTF-8"));
}
}
}); //close();
return response.take();
} public void close() throws IOException {
connection.close();
} }

RPCClient.java

  创建RPC测试类:

 import java.io.IOException;
import java.util.concurrent.TimeoutException; public class RPCTest { public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
RPCClient rpcClient = new RPCClient();
System.out.println(rpcClient.call("2"));
}
}

RPCTest

完整示例https://gitee.com/lfalex/rabbitmq-learn

参考官方文档:http://www.rabbitmq.com/getstarted.html

上一篇:阿里云OSS同城冗余存储正式商业化,提供云上同城容灾能力


下一篇:collect_list/collect_set(列转行)