1、工具类
package utils;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
/**
* Created by zjl
* 2020/10/23
**/
public class RabbitMQUtils {
public static ConnectionFactory connectionFactory;
static {
connectionFactory = new ConnectionFactory();
//设置连接rabbitmq主机
connectionFactory.setHost("127.0.0.1");
//设置连接rabbitmq端口
connectionFactory.setPort(5672);
//设置连接哪个虚拟主机
connectionFactory.setVirtualHost("/ems");
//设置访问虚拟主机的用户名
connectionFactory.setUsername("guest");
//设置访问虚拟主机的密码
connectionFactory.setPassword("guest");
}
public static Connection getConnection(){//创建链接mq的连接工厂对象
try{
//重量级资源,没必要每拿一次连接创建一次,这样的话代价有点大,所以我们在类加载的时候创建出来,只创建一次
// ConnectionFactory connectionFactory = new ConnectionFactory();
//设置连接rabbitmq主机
// connectionFactory.setHost("127.0.0.1");
// //设置连接rabbitmq端口
// connectionFactory.setPort(5672);
// //设置连接哪个虚拟主机
// connectionFactory.setVirtualHost("/ems");
// //设置访问虚拟主机的用户名
// connectionFactory.setUsername("guest");
// //设置访问虚拟主机的密码
// connectionFactory.setPassword("guest");
//获取连接对象
return connectionFactory.newConnection();
}catch (Exception e){
e.printStackTrace();
}
return null;
}
//关闭通道和关闭连接的工具方法
public static void closeConnectionAndChanel(Channel channel,Connection connection){//创建链接mq的连接工厂对象
try{
if(channel!=null) channel.close();
if(connection!=null) connection.close();
}catch (Exception e){
e.printStackTrace();
}
}
}
2、生产者
package helloword;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import org.junit.Test;
import utils.RabbitMQUtils;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
/**
* Created by zjl
* 2020/10/21
**/
public class Provider {
//生产消息
@Test
public void testSendMessage() throws IOException, TimeoutException {
// //创建链接mq的连接工厂对象
// ConnectionFactory connectionFactory = new ConnectionFactory();
// //设置连接rabbitmq主机
// connectionFactory.setHost("127.0.0.1");
// //设置连接rabbitmq端口
// connectionFactory.setPort(5672);
// //设置连接哪个虚拟主机
// connectionFactory.setVirtualHost("/ems");
// //设置访问虚拟主机的用户名
// connectionFactory.setUsername("guest");
// //设置访问虚拟主机的密码
// connectionFactory.setPassword("guest");
//
// //获取连接对象
// Connection connection = connectionFactory.newConnection();
//通过工具类获取连接
Connection connection = RabbitMQUtils.getConnection();
//获取连接中通道
Channel channel = connection.createChannel();
//通道绑定对应消息队列
//参数1:队列名称,如果队列不存在自动创建
//参数2:用来定义队列特性是否要持久化 true 持久化队列 false 不持久化队列
//参数3:exclusive是否独占队列 true 独占 false 不独占
//参数4:autoDelete:是否在消费完之后自动删除队列 true 删除 false 不自动删除
//参数5:额外附加参数
channel.queueDeclare("hello",false,false,false,null);
//发布消息
//参数1:交换机名称 参数2:队列名称 参数3:传递消息额外设置 参数4:消息的具体内容
channel.basicPublish("","hello",null,"hello rabbitmq".getBytes());
// channel.close();
// connection.close();
//通过工具类来关闭连接
RabbitMQUtils.closeConnectionAndChanel(channel,connection);
}
}
2、消费者
package helloword;
import com.rabbitmq.client.*;
import org.junit.Test;
import utils.RabbitMQUtils;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
/**
* Created by zjl
* 2020/10/23
**/
public class Customer {
public static void main(String[] args) throws IOException, TimeoutException {
// //创建链接mq的连接工厂对象
// ConnectionFactory connectionFactory = new ConnectionFactory();
// //设置连接rabbitmq主机
// connectionFactory.setHost("127.0.0.1");
// //设置连接rabbitmq端口
// connectionFactory.setPort(5672);
// //设置连接哪个虚拟主机
// connectionFactory.setVirtualHost("/ems");
// //设置访问虚拟主机的用户名
// connectionFactory.setUsername("guest");
// //设置访问虚拟主机的密码
// connectionFactory.setPassword("guest");
//
// //获取连接对象
// Connection connection = connectionFactory.newConnection();
//通过工具类获取连接
Connection connection = RabbitMQUtils.getConnection();
//获取连接中通道
Channel channel = connection.createChannel();
//通道绑定对应消息队列
//参数1:队列名称,如果队列不存在自动创建
//参数2:用来定义队列特性是否要持久化 true 持久化队列 false 不持久化队列
//参数3:exclusive是否独占队列 true 独占 false 不独占
//参数4:autoDelete:是否在消费完之后自动删除队列 true 删除 false 不自动删除
//参数5:额外附加参数
channel.queueDeclare("hello",false,false,false,null);
//发布消息
//参数1:消费哪个队列的消息 队列名称 参数2:开始消息的自动确认机制 参数3:消费时的回调接口
channel.basicConsume("hello",true,new DefaultConsumer(channel){
@Override //最后一个参数,消息队列中取出的消息
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
System.out.println("new String(body)=" + new String(body));
}
});
// 不建议关闭通道,因为要一直监听,而且很可能通道关闭了,回调函数还没来得及调用
// channel.close();
// connection.close();
}
}
4、测试结果
版权声明:本文为qq_41347385原创文章,遵循CC 4.0 BY-SA版权协议,转载请附上原文出处链接和本声明。