RabbitMq简单模式下生产者与消费者的实现
首先需要导入对应的依赖
<dependencies>
<dependency>
<groupId>com.rabbitmq</groupId>
<artifactId>amqp-client</artifactId>
<version>5.6.0</version>
</dependency>
</dependencies>
依赖导入完毕后,首先实现生产者
public static void main(String[] args) throws IOException, TimeoutException {
//1.创建连接工厂
final ConnectionFactory factory = new ConnectionFactory();
//2.设置参数
factory.setHost("127.0.0.1");
factory.setPort(5672);
factory.setVirtualHost("admin");
factory.setUsername("admin");
factory.setPassword("admin");
//3.创建连接connection
final Connection connection = factory.newConnection();
//4.创建channel
final Channel channel = connection.createChannel();
//创建队列
/**
* String queue(队列名称), boolean durable(是否持久化),
* boolean exclusive(1:是否独占,2:当connection关闭时是否删除队列),
* boolean autoDelete(是否自动删除,当没有consumer时是否自动删除),
* Map<String, Object> arguments 参数
*
*
*
* 如果没有hello_world的队列 则会创建队列 。如果有则不会创建
*/
channel.queueDeclare("hello_world",true,false,false,null);
//发送消息
/**
* String exchange:交换机
* String routingKey:路由名称
* BasicProperties props:配置信息
* byte[] body:发送消息数据
*/
String body = "hello consumer";
channel.basicPublish("","hello_world",null,body.getBytes());
channel.close();
connection.close();
}
其次:实现消费者
public static void main(String[] args) throws IOException, TimeoutException {
//1.创建连接工厂
final ConnectionFactory factory = new ConnectionFactory();
//2.设置参数
factory.setHost("127.0.0.1");
factory.setPort(5672);
factory.setVirtualHost("admin");
factory.setUsername("admin");
factory.setPassword("admin");
//3.创建连接connection
final Connection connection = factory.newConnection();
//4.创建channel
final Channel channel = connection.createChannel();
//创建队列
/**
* String queue(队列名称), boolean durable(是否持久化),
* boolean exclusive(1:是否独占,2:当connection关闭时是否删除队列),
* boolean autoDelete(是否自动删除,当没有consumer时是否自动删除),
* Map<String, Object> arguments 参数
*
*
*
* 如果没有hello_world的队列 则会创建队列 。如果有则不会创建
*/
channel.queueDeclare("hello_world",true,false,false,null);
channel.basicConsume("hello_world",true,new DefaultConsumer(channel) {
//回调方法
/**
*
* @param consumerTag 标识
* @param envelope 获取一些信息 交换机信息 路由key等信息
* @param properties 配置信息
* @param body 消息
* @throws IOException
*/
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
// super.handleDelivery(consumerTag, envelope, properties, body);
System.out.println("consumerTag==========》》》》"+consumerTag);
System.out.println("Exchange=========>>>>>>>"+envelope.getExchange()+"=====rount======>>>"+envelope.getRoutingKey());
System.out.println("properties================>>>"+properties.getMessageId());
System.out.println("body=============>>>>>>>>>"+ new String(body));
}
});
}
首先启动生产者将消息发送到mq,接着启动消费者读取消息 至此,rabbitMq简单模式就实现了
|