java-rabbitmq-实例pull模式拉取消息

package com.example.tutorials;

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

import java.util.HashMap;
import java.util.Map;
import java.util.Scanner;

/**
* 使用 channel.basicGet() 方法，拉取指定队列中的内容
* @create 2017-09-05
* amqp-client 4.2.0
**/
public class D7_PullSend {
private final static String QUEUE_NAME = "pull_queue";

/**
* 生产者，
* @param argv
* @throws Exception
*/
public static void main(String[] argv) throws Exception {
ConnectionFactory factory = new ConnectionFactory();
//设置登录账号
factory.setHost(ServerInfo.host);
factory.setPort(ServerInfo.port);
factory.setUsername(ServerInfo.uName);
factory.setPassword(ServerInfo.uPwd);
//链接服务器
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();

Map<String,Object> args = new HashMap<>();
//        args.put("x-message-ttl",15*1000);//消息过期时间为 15 秒
//args.put("x-expires",1*60*1000); //队列过期时间为 1 分钟
//定义一个队列
boolean duiable=false;//持久化
boolean exclusive = false;//排他队列
boolean autoDelete=false;//没有consumer时，队列是否自动删除
channel.queueDeclare(QUEUE_NAME, duiable, exclusive, autoDelete, args);

//发送消息
System.out.println("输入要发送的消息，退出输入 x ");
String message ;
//
AMQP.BasicProperties prop = new AMQP.BasicProperties
.Builder()
.expiration("15000") //消息过期时间为 15 秒
.build();
do{
Scanner scanner = new Scanner(System.in);
message = scanner.next();
channel.basicPublish(""
, QUEUE_NAME
, prop
, message.getBytes("UTF-8"));
System.out.println(" [x] Sent '" + message + "'");
}while(!"x".equals(message));
//关闭链接
channel.close();
connection.close();
}
}

package com.example.tutorials;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.GetResponse;

import java.util.Scanner;

/**
* 使用 channel.basicGet() 方法，拉取指定队列中的内容
* @create 2017-09-05
* amqp-client 4.2.0
**/
public class D7_PullRecv {
private final static String QUEUE_NAME = "pull_queue";

/**
* 消费者， "Hello World!"
* @param argv
* @throws Exception
*/
public static void main(String[] argv) throws Exception {
ConnectionFactory factory = new ConnectionFactory();
//设置登录账号
factory.setHost(ServerInfo.host);
factory.setPort(ServerInfo.port);
factory.setUsername(ServerInfo.uName);
factory.setPassword(ServerInfo.uPwd);
//链接服务器
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();
//定义一个队列
boolean duiable=false;//持久化
boolean exclusive = false;//排他队列
boolean autoDelete=false;//没有consumer时，队列是否自动删除
channel.queueDeclare(QUEUE_NAME, duiable, exclusive, autoDelete, null);

//接收消息
System.out.println(" [*] Waiting for messages. To exit press CTRL+C");
boolean autoAck = true; //自动应答

System.out.println("输入回车获取消息，退出输入 x ");
String message ="";
GetResponse resp ;
do{
Scanner scanner = new Scanner(System.in);
scanner.nextLine();
resp = channel.basicGet(QUEUE_NAME,autoAck);
if(resp==null){
System.out.println(QUEUE_NAME+" 队列无消息");
continue;
}
message = new String(resp.getBody(), "UTF-8");
System.out.println(String.format(" [x] Recv Count %s , msg = %s;"
,resp.getMessageCount()
,message));
}while(!"x".equals(message));
}
}