RabbitMQ工作模式1 Hello world工作模式
使用Java代码操作RabbitMQ
需求:使用简单模式完成消息传递,简单模式如下所示,P代表生产者,C代表消费者,中间是队列
六种模式中最简单的就是它(工作模式)
模式说明
实现步骤(接下来的工作模式也会安装相同的代码进行修改)
1 创建两个工程,生产者(P)和消费者(C)
创建一个新的项目
在项目中创建两个新的模块 分别为生产者和消费者
2 分别引入对应工程的依赖
他俩的pom分别引入
<dependencies> <!--rabbitMQ依赖 Java客户端--> <dependency> <groupId>com.rabbitmq</groupId> <artifactId>amqp-client</artifactId> <version>5.6.0</version> </dependency> </dependencies> <!--编译插件 --> <build> <plugins> <plugin> <groupId>org.apache.maven.plugins</groupId> <artifactId>maven-compiler-plugin</artifactId> <version>3.8.0</version> <configuration> <source>1.8</source> <target>1.8</target> </configuration> </plugin> </plugins> </build>
3 编写生产者发送消息
package com.wyh.producer; /** * @program: SpringBoot-RabbitMQ * @description: RabbitMQ生产者 * @author: 魏一鹤 * @createDate: 2022-03-23 22:40 **/ import com.rabbitmq.client.AMQP; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConnectionFactory; import java.io.IOException; import java.util.concurrent.TimeoutException; /** * producer主要用来发送消息 * * **/ public class HelloWorld { public static void main(String[] args) throws IOException, TimeoutException { //1.建立工厂 一般连接都是通过连接工厂进行连接 所以要创建连接工厂 ConnectionFactory factory=new ConnectionFactory(); //2 设置参数 比如说虚拟机 用户名 ip 密码 端口等 当然不设置也有默认参数 factory.setHost("127.0.0.1 ");//设置主机ip 172.16.98.133是远程的服务 就是rabbitMQ服务页面的ip 如果不设置默认值为localhost (127.0.0.1) factory.setPort(5672);//设置端口 默认值也是5672 factory.setVirtualHost("/itcast_wyh");//设置虚拟机 默认值/ 杠 factory.setUsername("weiyihe");//设置用户名 默认值 guest 游客 factory.setPassword("weiyihe");//设置密码 默认值 guest 游客 //3 创建连接Connection Connection connection = factory.newConnection(); //4 创建channel Channel channel = connection.createChannel(); //5 创建队列Queue 它的参数比较多 下面一一说明 /** * 1 String queue, 队列名称 * 2 boolean durable, 是否持久化,当MQ重启之后还在 持久化到数据库中 * 3 boolean exclusive, 它有两个功能 1是否独占:只能有一个消费者监听这个队列 2当connection关闭时,是否删除队列.一般设置为false * 4 boolean autoDelete, 是否自动删除,让没有consumer时,会自动删除掉 * 5 Map<String, Object> arguments 参数信息 * **/ //如果没有一个名字叫hello_world的队列,则会创建该队列,如果有则不会创建 AMQP.Queue.DeclareOk queue = channel.queueDeclare("hello_world", true, false, false, null); //6 发送消息 它的参数比较多 下面一一说明 //发送的消息队列 用来发送给消费者 一般发送的消息都是字节 String body="hello RabbitMQ!"; // 1 String exchange,交换机的名称,简单模式下交换机会使用默认的 "" // 2 String routingKey, 路由名称 // 3 BasicProperties props, 配置信息 // 4 byte[] body 真实的发送的消息数据 channel.basicPublish("","hello_world",null,body.getBytes()); //7 释放资源 channel.close(); connection.close(); } }
4 编写消费者接收消息
package com.wyh.consumer; import com.rabbitmq.client.*; import java.io.IOException; import java.util.concurrent.TimeoutException; /** * @program: SpringBoot-RabbitMQ * @description: RabbitMQ消费者Consumer * @author: 魏一鹤 * @createDate: 2022-03-24 22:08 **/ /** * consumer主要用来消费消息 * * **/ public class HelloWorld_consumer { public static void main(String[] args) throws IOException, TimeoutException { //1.建立工厂 一般连接都是通过连接工厂进行连接 所以要创建连接工厂 ConnectionFactory factory=new ConnectionFactory(); //2 设置参数 比如说虚拟机 用户名 ip 密码 端口等 当然不设置也有默认参数 factory.setHost("127.0.0.1");//设置主机ip 172.16.98.133是远程的服务 就是rabbitMQ服务页面的ip 如果不设置默认值为localhost (127.0.0.1) factory.setPort(5672);//设置端口 默认值也是5672 factory.setVirtualHost("/itcast_wyh");//设置虚拟机 默认值/ 杠 factory.setUsername("weiyihe");//设置用户名 默认值 guest 游客 factory.setPassword("weiyihe");//设置密码 默认值 guest 游客 //3 创建连接Connection Connection connection = factory.newConnection(); //4 创建channel Channel channel = connection.createChannel(); //5 创建队列Queue 它的参数比较多 下面一一说明 /** * 1 String queue, 队列名称 * 2 boolean durable, 是否持久化,当MQ重启之后还在 持久化到数据库中 * 3 boolean exclusive, 它有两个功能 1是否独占:只能有一个消费者监听这个队列 2当connection关闭时,是否删除队列.一般设置为false * 4 boolean autoDelete, 是否自动删除,让没有consumer时,会自动删除掉 * 5 Map<String, Object> arguments 参数信息 * **/ //如果没有一个名字叫hello_world的队列,则会创建该队列,如果有则不会创建 AMQP.Queue.DeclareOk queue = channel.queueDeclare("hello_world", true, false, false, null); //6 接收消息 它的参数比较多 下面一一说明 // String queue, 队列名称 // boolean autoAck, 是否自动确认 消费者收到消息会自动告诉MQ它收到了消息 // Consumer callback 回调对象 可以监听一些方法 //consumer本质是一个接口 需要创建它的实现类 Consumer consumer=new DefaultConsumer(channel){ //匿名内部类 重写它的方法 //回调方法 当收到消息后,会自动执行该方法 它有一些参数 //String consumerTag 标识 //Envelope envelope 可以获取一些信息 比如交换机 路由key //AMQP.BasicProperties properties 配置信息 // byte[] body 数据 // @Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { System.out.println("consumerTag = " + consumerTag); System.out.println("exchange = " + envelope.getExchange()); System.out.println("routingKey = " + envelope.getRoutingKey()); System.out.println("properties = " + properties); System.out.println("body = " + new String(body)); } /** 打印的信息 consumerTag = amq.ctag-7ojIfmsjb9wcpbkwzwEobw exchange = routingKey = hello_world properties = #contentHeader<basic>(content-type=null, content-encoding=null, headers=null, delivery-mode=null, priority=null, correlation-id=null, reply-to=null, expiration=null, message-id=null, timestamp=null, type=null, user-id=null, app-id=null, cluster-id=null) body = hello RabbitMQ! consumerTag = amq.ctag-7ojIfmsjb9wcpbkwzwEobw exchange = routingKey = hello_world properties = #contentHeader<basic>(content-type=null, content-encoding=null, headers=null, delivery-mode=null, priority=null, correlation-id=null, reply-to=null, expiration=null, message-id=null, timestamp=null, type=null, user-id=null, app-id=null, cluster-id=null) body = hello RabbitMQ 222! **/ }; channel.basicConsume("hello_world",true,consumer); //消费者本质是一个监听 所以不要去关闭资源 } }
consumerTag = amq.ctag-7ojIfmsjb9wcpbkwzwEobw
exchange =
routingKey = hello_world
properties = #contentHeader<basic>(content-type=null, content-encoding=null, headers=null, delivery-mode=null, priority=null, correlation-id=null, reply-to=null, expiration=null, message-id=null, timestamp=null, type=null, user-id=null, app-id=null, cluster-id=null)
body = hello RabbitMQ!
consumerTag = amq.ctag-7ojIfmsjb9wcpbkwzwEobw
exchange =
routingKey = hello_world
properties = #contentHeader<basic>(content-type=null, content-encoding=null, headers=null, delivery-mode=null, priority=null, correlation-id=null, reply-to=null, expiration=null, message-id=null, timestamp=null, type=null, user-id=null, app-id=null, cluster-id=null)
body = hello RabbitMQ 222!
5 测试
现在是没有队列的
运行程序代码,发现有了一个队列
package com.wyh.producer; /** * @program: SpringBoot-RabbitMQ * @description: RabbitMQ生产者 * @author: 魏一鹤 * @createDate: 2022-03-23 22:40 **/ import com.rabbitmq.client.AMQP; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConnectionFactory; import java.io.IOException; import java.util.concurrent.TimeoutException; /** * producer主要用来发送消息 * * **/ public class HelloWorld { public static void main(String[] args) throws IOException, TimeoutException { //1.建立工厂 一般连接都是通过连接工厂进行连接 所以要创建连接工厂 ConnectionFactory factory=new ConnectionFactory(); //2 设置参数 比如说虚拟机 用户名 ip 密码 端口等 当然不设置也有默认参数 factory.setHost("127.0.0.1");//设置主机ip 172.16.98.133是远程的服务 就是rabbitMQ服务页面的ip 如果不设置默认值为localhost (127.0.0.1) factory.setPort(5672);//设置端口 默认值也是5672 factory.setVirtualHost("/itcast_wyh");//设置虚拟机 默认值/ 杠 factory.setUsername("weiyihe");//设置用户名 默认值 guest 游客 factory.setPassword("weiyihe");//设置密码 默认值 guest 游客 //3 创建连接Connection Connection connection = factory.newConnection(); //4 创建channel Channel channel = connection.createChannel(); //5 创建队列Queue 它的参数比较多 下面一一说明 /** * 1 String queue, 队列名称 * 2 boolean durable, 是否持久化,当MQ重启之后还在 持久化到数据库中 * 3 boolean exclusive, 它有两个功能 1是否独占:只能有一个消费者监听这个队列 2当connection关闭时,是否删除队列.一般设置为false * 4 boolean autoDelete, 是否自动删除,让没有consumer时,会自动删除掉 * 5 Map<String, Object> arguments 参数信息 * **/ //如果没有一个名字叫hello_world的队列,则会创建该队列,如果有则不会创建 AMQP.Queue.DeclareOk queue = channel.queueDeclare("hello_world", true, false, false, null); //6 发送消息 它的参数比较多 下面一一说明 //发送的消息队列 用来发送给消费者 一般发送的消息都是字节 String body="hello RabbitMQ!"; // 1 String exchange,交换机的名称,简单模式下交换机会使用默认的 "" // 2 String routingKey, 路由名称 // 3 BasicProperties props, 配置信息 // 4 byte[] body 真实的发送的消息数据 channel.basicPublish("","hello_world",null,body.getBytes()); //7 释放资源 //channel.close(); //connection.close(); } }
可以发现,连接和channel都有了,而且队列多了一条信息,因为我们又执行了一次