public class Send {
public static void main(String[] args) throws Exception{
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();
/*获取tcp其中的一个通道后 进行队列声明
参数1 队列名字
参数2 是否持久化
参数3 只被一个链接使用，链接关闭后删除
参数4 最后一个消费者退订后删除*/
channel.queueDeclare("wuqidi", false, false, false, null);
/*发布到队列中
* 参数1 交换机 默认交换机
* 参数2 路由key
* 参数3 参数
* 参数4 发送的信息 proCon
* */
channel.basicPublish("", "wuqidi", null, "十九大重要会议在即。".getBytes());
channel.close();
connection.close();
}
}
public class Receive {
public static void main(String[] args) throws Exception{
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();
/*获取tcp其中的一个通道后 进行队列声明
参数1 队列名字
参数2 是否持久化
参数3 只被一个链接使用，链接关闭后删除
参数4 最后一个消费者退订后删除*/
channel.queueDeclare("wuqidi", false, false, false, null);
//callback
com.rabbitmq.client.Consumer callback = new Consumer() {
@Override
public void handleShutdownSignal(String consumerTag,
ShutdownSignalException sig) {
// TODO Auto-generated method stub
System.out.println("关闭信号源" + consumerTag);
}
@Override
public void handleRecoverOk(String consumerTag) {
// TODO Auto-generated method stub
System.out.println("返回ok" + consumerTag);
}
@Override
public void handleDelivery(String arg0, Envelope arg1,
BasicProperties arg2, byte[] body) throws IOException {
// TODO Auto-generated method stub
System.out.println(arg0 +"---" + new String(body, "utf-8"));
}
@Override
public void handleConsumeOk(String consumerTag) {
System.out.println("返回123ok" + consumerTag);
}
@Override
public void handleCancelOk(String consumerTag) {
System.out.println("信号源ok" + consumerTag);
}
@Override
public void handleCancel(String consumerTag) throws IOException {
System.out.println("返回consumerTag" + consumerTag);
}
};
channel.basicConsume("wuqidi", callback);
}
}
后期如果用spring搭建框架，可以在xml中进行声明queue，包括绑定，监听等。服务层中只需要获取amqptemplate然后发送或者接受即可。