一、简单的发送与接收消息 HelloWorld
1. 发送消息
发送消息首先要获取与rabbitmq-server的连接，然后从渠道(chann)中指定的queue发送消息 ， 不能定义两个queue名字相同，但属性不同
示例：
<code>package com.zf.rabbitmq01;
import java.io.IOException;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
/**
* 发送消息
* @author zhanghuan
*
*/
public class Sender01 {
public static void main(String[] args) throws IOException {
ConnectionFactory connFac = new ConnectionFactory() ;
//RabbitMQ-Server安装在本机，所以直接用127.0.0.1
connFac.setHost("127.0.0.1");
//创建一个连接
Connection conn = connFac.newConnection() ;
//创建一个渠道
Channel channel = conn.createChannel() ;
//定义Queue名称
String queueName = "queue01" ;
//为channel定义queue的属性，queueName为Queue名称
channel.queueDeclare( queueName , false, false, false, null) ;
String msg = "Hello World!";
//发送消息
channel.basicPublish("", queueName , null , msg.getBytes());
System.out.println("send message[" + msg + "] to "+ queueName +" success!");
channel.close();
conn.close();
}
}
/<code>
<code>package com.zf.rabbitmq01;
import java.io.IOException;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.ConsumerCancelledException;
import com.rabbitmq.client.QueueingConsumer;
import com.rabbitmq.client.QueueingConsumer.Delivery;
import com.rabbitmq.client.ShutdownSignalException;
/**
* 接收消息
* @author zhanghuan
*
*/
public class Recv01 {
public static void main(String[] args) throws IOException, ShutdownSignalException, ConsumerCancelledException, InterruptedException {
ConnectionFactory connFac = new ConnectionFactory() ;
connFac.setHost("127.0.0.1");
Connection conn = connFac.newConnection() ;
Channel channel = conn.createChannel() ;
String queueName = "queue01";
channel.queueDeclare(queueName, false, false, false, null) ;
//上面的部分，与Sender01是一样的
//配置好获取消息的方式
QueueingConsumer consumer = new QueueingConsumer(channel) ;
channel.basicConsume(queueName, true, consumer) ;
//循环获取消息
while(true){
//获取消息，如果没有消息，这一步将会一直阻塞
Delivery delivery = consumer.nextDelivery() ;
String msg = new String(delivery.getBody()) ;
System.out.println("received message[" + msg + "] from " + queueName);
}
}
}
/<code>
二、消息确认与公平调度消费者
从本节开始称Sender为生产者 ， Recv为消费者
1. 消息确认
为了确保消息一定被消费者处理，rabbitMQ提供了消息确认功能，就是在消费者处理完任务之后，就给服务器一个回馈，服务器就会将该消息删除，如果消费者超时不回馈，那么服务器将就将该消息重新发送给其他消费者默认是开启的，在消费者端通过下面的方式开启消息确认, 首先将autoAck自动确认关闭，等我们的任务执行完成之后，手动的去确认，类似JDBC的autocommit一样
<code>QueueingConsumer consumer = new QueueingConsumer(channel);
Boolean autoAck = false;
channel.basicConsume("hello", autoAck, consumer);
/<code>
在前面的例子中使用的是channel.basicConsume(channelName, true, consumer) ; 在接收到消息后，就会自动反馈一个消息给服务器。
下面这个例子来测试消息确认的功能。
<code>package com.zf.rabbitmq03;
import java.io.IOException;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
/**
* 发送消息
* @author zhanghuan
*
*/
public class Sender03 {
public static void main(String[] args) throws IOException {
ConnectionFactory connFac = new ConnectionFactory() ;
//RabbitMQ-Server安装在本机，所以直接用127.0.0.1
connFac.setHost("127.0.0.1");
//创建一个连接
Connection conn = connFac.newConnection() ;
//创建一个渠道
Channel channel = conn.createChannel() ;
//定义Queue名称
String queueName = "queue01" ;
//为channel定义queue的属性，queueName为Queue名称
channel.queueDeclare( queueName , false, false, false, null) ;
String msg = "Hello World!";
//发送消息
channel.basicPublish("", queueName , null , msg.getBytes());
System.out.println("send message[" + msg + "] to "+ queueName +" success!");
channel.close();
conn.close();
}
}
/<code>
与Sender01.java一样，没有什么区别。
<code>package com.zf.rabbitmq03;
import java.io.IOException;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.ConsumerCancelledException;
import com.rabbitmq.client.QueueingConsumer;
import com.rabbitmq.client.QueueingConsumer.Delivery;
import com.rabbitmq.client.ShutdownSignalException;
/**
* 接收消息
* @author zhanghuan
*
*/
public class Recv03 {
public static void main(String[] args) throws IOException, ShutdownSignalException, ConsumerCancelledException, InterruptedException {
ConnectionFactory connFac = new ConnectionFactory() ;
connFac.setHost("127.0.0.1");
Connection conn = connFac.newConnection() ;
Channel channel = conn.createChannel() ;
String channelName = "channel01";
channel.queueDeclare(channelName, false, false, false, null) ;
//配置好获取消息的方式
QueueingConsumer consumer = new QueueingConsumer(channel) ;
//取消 autoAck
Boolean autoAck = false ;
channel.basicConsume(channelName, autoAck, consumer) ;
//循环获取消息
while(true){
//获取消息，如果没有消息，这一步将会一直阻塞
Delivery delivery = consumer.nextDelivery() ;
String msg = new String(delivery.getBody()) ;
//确认消息，已经收到
channel.basicAck(delivery.getEnvelope().getDeliveryTag()
, false);
System.out.println("received message[" + msg + "] from " + channelName);
}
}
}
/<code>
注意：一旦将autoAck关闭之后，一定要记得处理完消息之后，向服务器确认消息。否则服务器将会一直转发该消息如果将上面的channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);注释掉， Sender03.java只需要运行一次 ， Recv03.java每次运行将都会收到HelloWorld消息
注意：但是这样还是不够的，如果rabbitMQ-Server突然挂掉了，那么还没有被读取的消息还是会丢失 ，所以我们可以让消息持久化。 只需要在定义Queue时，设置持久化消息就可以了，方法如下：
<code>boolean durable = true;channel.queueDeclare(channelName, durable, false, false, null);
/<code>
这样设置之后，服务器收到消息后就会立刻将消息写入到硬盘，就可以防止突然服务器挂掉，而引起的数据丢失了。 但是服务器如果刚收到消息，还没来得及写入到硬盘，就挂掉了，这样还是无法避免消息的丢失。
2. 公平调度
上一个例子能够实现发送一个Message与接收一个Message
从上一个Recv01中可以看出，必须处理完一个消息，才会去接收下一个消息。如果生产者众多，那么一个消费者肯定是忙不过来的。此时就可以用多个消费者来对同一个Channel的消息进行处理，并且要公平的分配任务给多个消费者。不能部分很忙，部分总是空闲
实现公平调度的方式就是让每个消费者在同一时刻会分配一个任务。 通过channel.basicQos(1);可以设置
列如：
当有多个消费者同时收取消息，且每个消费者在接收消息的同时，还要做其它的事情，且会消耗很长的时间，在此过程中可能会出现一些意外，比如消息接收到一半的时候，一个消费者宕掉了，这时候就要使用消息接收确认机制，可以让其它的消费者再次执行刚才宕掉的消费者没有完成的事情。另外，在默认情况下，我们创建的消息队列以及存放在队列里面的消息，都是非持久化的，也就是说当RabbitMQ宕掉了或者是重启了，创建的消息队列以及消息都不会保存，为了解决这种情况，保证消息传输的可靠性，我们可以使用RabbitMQ提供的消息队列的持久化机制。
生产者：
<code>import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.MessageProperties;
public class ClientSend1 {
public static final String queue_name="my_queue";
public static final Boolean durable=true;
//消息队列持久化
public static void main(String[] args)
throws java.io.IOException{
ConnectionFactory factory=new ConnectionFactory();
//创建连接工厂
factory.setHost("localhost");
factory.setVirtualHost("my_mq");
factory.setUsername("zhxia");
factory.setPassword("123456");
Connection connection=factory.newConnection();
//创建连接
Channel channel=connection.createChannel();
//创建信道
channel.queueDeclare(queue_name, durable, false, false, null);
//声明消息队列，且为可持久化的
String message="Hello world"+Math.random();
//将队列设置为持久化之后，还需要将消息也设为可持久化的，MessageProperties.PERSISTENT_TEXT_PLAIN
channel.basicPublish("", queue_name, MessageProperties.PERSISTENT_TEXT_PLAIN,message.getBytes());
System.out.println("Send message:"+message);
channel.close();
connection.close();
}
}
/<code>
说明：行17 和行20 需要同时设置，也就是将队列设置为持久化之后，还需要将发送的消息也要设置为持久化才能保证队列和消息一直存在
消费者：
<code>import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.QueueingConsumer;
public class ClientReceive1 {
public static final String queue_name="my_queue";
public static final Boolean autoAck=false;
public static final Boolean durable=true;
public static void main(String[] args)
throws java.io.IOException,java.lang.InterruptedException{
ConnectionFactory factory=new ConnectionFactory();
factory.setHost("localhost");
factory.setVirtualHost("my_mq");
factory.setUsername("zhxia");
factory.setPassword("123456");
Connection connection=factory.newConnection();
Channel channel=connection.createChannel();
channel.queueDeclare(queue_name, durable, false, false, null);
System.out.println("Wait for message");
channel.basicQos(1);
//消息分发处理
QueueingConsumer consumer=new QueueingConsumer(channel);
channel.basicConsume(queue_name, autoAck, consumer);
while(true){
Thread.sleep(500);
QueueingConsumer.Delivery deliver=consumer.nextDelivery();
String message=new String(deliver.getBody());
System.out.println("Message received:"+message);
channel.basicAck(deliver.getEnvelope().getDeliveryTag(), false);
}
}
}
/<code>
说明：行22： 设置RabbitMQ调度分发消息的方式，也就是告诉RabbitMQ每次只给消费者处理一条消息，也就是等待消费者处理完并且已经对刚才处理的消息进行确认之后， 才发送下一条消息，防止消费者太过于忙碌。如下图所示：
三、发布/订阅消息
前面都是一条消息只会被一个消费者处理。
如果要每个消费者都处理同一个消息，rabbitMq也提供了相应的方法。
在以前的程序中，不管是生产者端还是消费者端都必须知道一个指定的QueueName才能发送、获取消息。 而rabbitMQ消息模型的核心思想是生产者不会将消息直接发送给队列。
因为，生产者通常不会知道消息将会被哪些消费者接收。
生产者的消息虽然不是直接发送给Queue，但是消息会交给Exchange，所以需要定义Exchange的消息分发模式 ，之前的程序中，有如下一行代码：
<code>channel.basicPublish("", queueName , null , msg.getBytes());
/<code>第一个参数为空字符串，其实第一个参数就是ExchangeName，这里用空字符串，就表示消息会交给默认的Exchange。
下面我们将自己定义Exchange的属性。
<code>package com.zf.rabbitmq04;
import java.io.IOException;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
/**
* 发送消息
* @author zhanghuan
*
*/
public class Sender04 {
public static void main(String[] args) throws IOException {
ConnectionFactory connFac = new ConnectionFactory() ;
//RabbitMQ-Server安装在本机，所以直接用127.0.0.1
connFac.setHost("127.0.0.1");
//创建一个连接
Connection conn = connFac.newConnection() ;
//创建一个渠道
Channel channel = conn.createChannel() ;
//定义ExchangeName,第二个参数是Exchange的类型，fanout表示消息将会分列发送给多账户
String exchangeName = "news" ;
channel.exchangeDeclare(exchangeName, "fanout") ;
String msg = "Hello World!";
//发送消息，这里与前面的不同，这里第一个参数不再是字符串，而是ExchangeName ，第二个参数也不再是queueName，而是空字符串
channel.basicPublish( exchangeName , "" , null , msg.getBytes());
System.out.println("send message[" + msg + "] to exchange "+ exchangeName +" success!");
channel.close();
conn.close();
}
}
/<code>
Send04.java 发送消息时没有指定的queueName 用的空字符串代替的。 Exchange的类型有direct, topic, headers 、 fanout四种，上面用的是fanout类型
<code>package com.zf.rabbitmq04;
import java.io.IOException;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.ConsumerCancelledException;
import com.rabbitmq.client.QueueingConsumer;
import com.rabbitmq.client.QueueingConsumer.Delivery;
import com.rabbitmq.client.ShutdownSignalException;
/**
* 接收消息
* @author zhanghuan
*
*/
public class Recv04_01 {
public static void main(String[] args) throws IOException, ShutdownSignalException, ConsumerCancelledException, InterruptedException {
ConnectionFactory connFac = new ConnectionFactory() ;
connFac.setHost("127.0.0.1");
Connection conn = connFac.newConnection() ;
Channel channel = conn.createChannel() ;
String exchangeName = "news" ;
channel.exchangeDeclare(exchangeName, "fanout") ;
//这里使用没有参数的queueDeclare方法创建Queue并获取QueueName
String queueName = channel.queueDeclare().getQueue() ;
//将queue绑定到Exchange中
channel.queueBind( queueName, exchangeName, "") ;
//配置好获取消息的方式
QueueingConsumer consumer = new QueueingConsumer(channel) ;
channel.basicConsume(queueName, true, consumer) ;
//循环获取消息
while(true){
//获取消息，如果没有消息，这一步将会一直阻塞
Delivery delivery = consumer.nextDelivery() ;
String msg = new String(delivery.getBody()) ;
System.out.println("received message[" + msg + "] from " + queueName);
}
}
}
/<code>
Recv04_01.java 使用channel.queueDeclare()方法创建了一个Queue，该Queue有系统创建，并分配了一个随机的名称。 然后将该Queue与与Exchange绑定在一起。 该Queue就能从Exchange中后去消息了。
测试
将Recv04_01.java 文件复制几份 Recv04_02.java Recv04_03.java，然后执行Recv04_01 与 Recv04_02，接下来执行Sender04发送消息，可以看到Recv04_01 与Recv04_02都接收到了消息。然后执行Recv04_03，没有获取到任何消息。接下来再执行Sender04发送消息，可以看到Recv04_01 、Recv04_02与Recv04_03都接收到了消息。
说明Exchange在收到生产者的消息后，会将消息发送给当前已经与它绑定了的所有Queue 。 然后被移除。
四、消息路由
生产者会生产出很多消息 ， 但是不同的消费者可能会有不同的需求，只需要接收指定的消息，其他的消息需要被过滤掉。 这时候就可以对消息进行过滤了。 在消费者端设置好需要接收的消息类型。
如果不使用默认的Exchange发送消息，而是使用我们自定定义的Exchange发送消息，那么下面这个方法的第二个参数就不是QueueName了，而是消息的类型。
<code>channel.basicPublish( exchangeName , messageType , null , msg.getBytes());
/<code>
示例：
<code>package com.zf.rabbitmq05;
import java.io.IOException;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
/**
* 发送消息
* @author zhanghuan
*
*/
public class Sender05 {
public static void main(String[] args) throws IOException {
ConnectionFactory connFac = new ConnectionFactory() ;
//RabbitMQ-Server安装在本机，所以直接用127.0.0.1
connFac.setHost("127.0.0.1");
//创建一个连接
Connection conn = connFac.newConnection() ;
//创建一个渠道
Channel channel = conn.createChannel() ;
String exchangeName = "exchange02";
String messageType = "type01";
channel.exchangeDeclare(exchangeName, "direct") ;
//定义Queue名
String msg = "Hello World!";
//发送消息
channel.basicPublish( exchangeName , messageType , null , msg.getBytes());
System.out.println("send message[" + msg + "] to "+ exchangeName +" success!");
channel.close();
conn.close();
}
}
/<code>
<code>package com.zf.rabbitmq05;
import java.io.IOException;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.ConsumerCancelledException;
import com.rabbitmq.client.QueueingConsumer;
import com.rabbitmq.client.QueueingConsumer.Delivery;
import com.rabbitmq.client.ShutdownSignalException;
/**
* 接收消息
* @author zhanghuan
*
*/
public class Recv05_01 {
public static void main(String[] args) throws IOException, ShutdownSignalException, ConsumerCancelledException, InterruptedException {
ConnectionFactory connFac = new ConnectionFactory() ;
connFac.setHost("127.0.0.1");
Connection conn = connFac.newConnection() ;
Channel channel = conn.createChannel() ;
String exchangeName = "exchange02";
channel.exchangeDeclare(exchangeName, "direct") ;
String queueName = channel.queueDeclare().getQueue() ;
//第三个参数就是type，这里表示只接收type01类型的消息。
channel.queueBind(queueName, exchangeName, "type01") ;
//也可以选择接收多种类型的消息。只需要再下面再绑定一次就可以了
channel.queueBind(queueName, exchangeName, "type02") ;
//配置好获取消息的方式
QueueingConsumer consumer = new QueueingConsumer(channel) ;
channel.basicConsume(queueName, true, consumer) ;
//循环获取消息
while(true){
//获取消息，如果没有消息，这一步将会一直阻塞
Delivery delivery = consumer.nextDelivery() ;
String msg = new String(delivery.getBody()) ;
System.out.println("received message[" + msg + "] from " + exchangeName);
}
}
}
/<code>
这时，启动Recv05_01.java 然后启动Sender05.java ，消费者端就会收到消息。然后将Sender05.java 中的messageType分别改为type02 type03 然后发送消息 ， 可以看到消费者端能接收到type02的消息，但是不能接收到type03的消息。
五、Topic类型消息
上一节中使用了消息路由，消费者可以选择性的接收消息。 但是这样还是不够灵活。
比如某个消费者要订阅娱乐新闻消息 。 包括新浪、网易、腾讯的娱乐新闻。那么消费者就需要绑定三次，分别绑定这三个网站的消息类型。 如果新闻门户更多了，那么消费者将要绑定个更多的消息类型， 其实消费者只是需要订阅娱乐新闻，不管是哪个网站的新闻，都需要。 那么在rabbitMQ中可以使用topic类型。 模糊匹配消息类型。
模糊匹配中的 *代表一个 #代表零个或1个
示例：
<code>package com.zf.rabbitmq06;
import java.io.IOException;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.ConsumerCancelledException;
import com.rabbitmq.client.QueueingConsumer;
import com.rabbitmq.client.QueueingConsumer.Delivery;
import com.rabbitmq.client.ShutdownSignalException;
/**
* 接收消息
* @author zhanghuan
*
*/
public class Recv06_01 {
public static void main(String[] args) throws IOException, ShutdownSignalException, ConsumerCancelledException, InterruptedException {
ConnectionFactory connFac = new ConnectionFactory() ;
connFac.setHost("127.0.0.1");
Connection conn = connFac.newConnection() ;
Channel channel = conn.createChannel() ;
String exchangeName = "exchange03";
channel.exchangeDeclare(exchangeName, "topic") ;
String queueName = channel.queueDeclare().getQueue() ;
//第三个参数就是type，这里表示只接收type01类型的消息。
channel.queueBind(queueName, exchangeName, "#.type01") ;
//配置好获取消息的方式
QueueingConsumer consumer = new QueueingConsumer(channel) ;
channel.basicConsume(queueName, true, consumer) ;
//循环获取消息
while(true){
//获取消息，如果没有消息，这一步将会一直阻塞
Delivery delivery = consumer.nextDelivery() ;
String msg = new String(delivery.getBody()) ;
System.out.println("received message[" + msg + "] from " + exchangeName);
}
}
}
/<code>
<code>package com.zf.rabbitmq06;
import java.io.IOException;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
/**
* 发送消息
* @author zhanghuan *
*/
public class Sender06 {
public static void main(String[] args) throws IOException {
ConnectionFactory connFac = new ConnectionFactory() ;
//RabbitMQ-Server安装在本机，所以直接用127.0.0.1
connFac.setHost("127.0.0.1");
//创建一个连接
Connection conn = connFac.newConnection() ;
//创建一个渠道
Channel channel = conn.createChannel() ;
String exchangeName = "exchange03";
String messageType = "fs.type01";
channel.exchangeDeclare(exchangeName, "topic") ;
//定义Queue名
String msg = "Hello World!";
//发送消息
channel.basicPublish( exchangeName , messageType , null , msg.getBytes());
System.out.println("send message[" + msg + "] to "+ exchangeName +" success!");
channel.close();
conn.close();
}
}
/<code>
使用topic之后 。不管Sender端发送的消息类型是fs.type01 还是 xx.type01 还是 type01 ，消费者都会收到消息
六、RPC 远程过程调用
当客户端想要调用服务器的某个方法来完成某项功能时，就可以使用rabbitMQ支持的PRC服务。
其实RPC服务与普通的收发消息的区别不大， RPC的过程其实就是客户端向服务端定义好的Queue发送消息，其中携带的消息就应该是服务端将要调用的方法的参数 ，并使用Propertis告诉服务端将结果返回到指定的Queue。
示例：
<code>package com.zf.rabbitmq07;
import java.io.IOException;
import com.rabbitmq.client.AMQP.BasicProperties;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.ConsumerCancelledException;
import com.rabbitmq.client.QueueingConsumer;
import com.rabbitmq.client.QueueingConsumer.Delivery;
import com.rabbitmq.client.ShutdownSignalException;
public class RPCServer {
public static final String RPC_QUEUE_NAME = "rpc_queue";
public static String sayHello(String name){
return "hello " + name ;
}
public static void main(String[] args) throws IOException, ShutdownSignalException, ConsumerCancelledException, InterruptedException {
ConnectionFactory connFac = new ConnectionFactory() ;
connFac.setHost("localhost");
Connection conn = connFac.newConnection() ;
Channel channel = conn.createChannel() ;
channel.queueDeclare(RPC_QUEUE_NAME, false, false, false, null) ;
QueueingConsumer consumer = new QueueingConsumer(channel);
channel.basicConsume(RPC_QUEUE_NAME, false , consumer) ;
while(true){
System.out.println("服务端等待接收消息..");
Delivery deliver = consumer.nextDelivery() ;
System.out.println("服务端成功收到消息..");
BasicProperties props = deliver.getProperties() ;
String message = new String(deliver.getBody() , "UTF-8") ;
String responseMessage = sayHello(message) ;
BasicProperties responseProps = new BasicProperties.Builder()
.correlationId(props.getCorrelationId())
.build() ;
//将结果返回到客户端Queue
channel.basicPublish("", props.getReplyTo() , responseProps , responseMessage.getBytes("UTF-8") ) ;
//向客户端确认消息
channel.basicAck(deliver.getEnvelope().getDeliveryTag(), false);
System.out.println("服务端返回消息完成..");
}
}
}
/<code>
<code>package com.zf.rabbitmq07;
import java.io.IOException;
import java.util.UUID;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.ConsumerCancelledException;
import com.rabbitmq.client.QueueingConsumer;
import com.rabbitmq.client.AMQP.BasicProperties;
import com.rabbitmq.client.QueueingConsumer.Delivery;
import com.rabbitmq.client.ShutdownSignalException;
public class RPCClient {
public static final String RPC_QUEUE_NAME = "rpc_queue";
public static void main(String[] args) throws IOException, ShutdownSignalException, ConsumerCancelledException, InterruptedException {
ConnectionFactory connFac = new ConnectionFactory() ;
connFac.setHost("localhost");
Connection conn = connFac.newConnection() ;
Channel channel = conn.createChannel() ;
//响应QueueName ，服务端将会把要返回的信息发送到该Queue
String responseQueue = channel.queueDeclare().getQueue() ;
String correlationId = UUID.randomUUID().toString() ;
BasicProperties props = new BasicProperties.Builder()
.replyTo(responseQueue)
.correlationId(correlationId)
.build();
String message = "is_zhoufeng";
channel.basicPublish( "" , RPC_QUEUE_NAME , props , message.getBytes("UTF-8"));
QueueingConsumer consumer = new QueueingConsumer(channel) ;
channel.basicConsume( responseQueue , consumer) ;
while(true){
Delivery delivery = consumer.nextDelivery() ;
if(delivery.getProperties().getCorrelationId().equals(correlationId)){
String result = new String(delivery.getBody()) ;
System.out.println(result);
}
}
}
}/<code>
写在最后：
- 针对于Java程序员，笔者最近整理了一些面试真题，思维导图，程序人生等PDF学习资料；
- 关注私信我"86"，即可获取！
- 希望读到这的您能点个小赞和关注下我，以后还会更新技术干货，谢谢您的支持！
分享到:
閱讀更多 Java架構師丨蘇先生 的文章
