1. 创建Maven工程
创建一个Maven工程
<dependencies><dependency><groupId>com.rabbitmq</groupId><artifactId>amqp-client</artifactId><version>5.10.0</version></dependency></dependencies>
2. 消息生产者
消息必须由生产者提供。
package com.zju.rabbitmq.simple;import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;/*** @Description* @create 2024-09-22 11:25*/
public class Producer {public static void main(String[] args) {/*** 所有的消息中间件技术都是基于tcp/ip协议基础之上构建新型的协议规范，子不过rabbitmq遵循的是amqp* ip：port** 1. 创建连接工程* 2. 创建连接Connection* 3. 通过连接获取通道Channel* 4. 通过通道创建交换机，声明队列，绑定关系，路由key，发送消息和接收消息* 5. 准备消息内容* 6. 发送消息给队列* 7. 关闭通道* 8. 关闭连接*/// 1. 创建连接工厂ConnectionFactory connectionFactory = new ConnectionFactory();// 2. 设置连接参数获得ConnectionconnectionFactory.setHost("127.0.0.1");connectionFactory.setPort(5672);connectionFactory.setVirtualHost("/");connectionFactory.setUsername("admin");connectionFactory.setPassword("admin");Connection connection = null;Channel channel = null;try{connection = connectionFactory.newConnection("生产者");// 3. 通过连接获取通道Channelchannel = connection.createChannel();// 4. 申明队列queue存储消息/** 如果队列不存在，则会创建* Rabbitmq不允许创建两个相同的队列名称，否则会报错。** @params1： queue 队列的名称* @params2： durable 队列是否持久化* @params3： exclusive 是否排他，即是否私有的，如果为true,会对当前队列加锁，其他的通道不能访问，并且连接自动关闭* @params4： autoDelete 是否自动删除，当最后一个消费者断开连接之后是否自动删除消息。* @params5： arguments 可以设置队列附加参数，设置队列的有效期，消息的最大长度，队列的消息生命周期等等。* */channel.queueDeclare("gzh_queue",false,false,false,null);// 5.准备发送消息的内容String message = "Hello world!";// 6.发送消息给中间件// @params1: 交换机exchange// @params2: 队列名称/routing// @params3: 属性配置// @params4: 发送消息的内容channel.basicPublish("","gzh_queue",null,message.getBytes());System.out.println("消息发送成功!");}catch (Exception ex){ex.printStackTrace();System.out.println("发送消息出现异常...");}finally {// 7: 关闭通道if (channel != null && channel.isOpen()) {try {channel.close();} catch (Exception ex) {ex.printStackTrace();}}// 8. 释放连接if (connection != null) {try {connection.close();} catch (Exception ex) {ex.printStackTrace();}}}}
}
3. 消息消费者
消息可以由消费者进行消费。
package com.zju.rabbitmq.simple;import java.io.IOException;import com.rabbitmq.client.*;/*** @author guozonghao* @Description* @create 2024-09-22 11:25*/
public class Consumer {public static void main(String[] args) {/*** 所有的消息中间件技术都是基于tcp/ip协议基础之上构建新型的协议规范，子不过rabbitmq遵循的是amqp* ip：port** 1. 创建连接工程* 2. 创建连接Connection* 3. 通过连接获取通道Channel* 4. 通过channel从队列中获取消息* 5. 关闭通道* 6. 关闭连接*/// 1. 创建连接工厂ConnectionFactory connectionFactory = new ConnectionFactory();// 2. 设置连接参数获得ConnectionconnectionFactory.setHost("47.98.238.134");connectionFactory.setPort(5672);connectionFactory.setVirtualHost("/");connectionFactory.setUsername("admin");connectionFactory.setPassword("gzh20001229.");Connection connection = null;Channel channel = null;try{connection = connectionFactory.newConnection("生产者");// 3. 通过连接获取通道Channelchannel = connection.createChannel();// 4. 通过channel从队列中获取消息channel.basicConsume("gzh_queue", true, new DeliverCallback() {public void handle(java.lang.String s, Delivery delivery) throws IOException {System.out.println("收到的消息是："+ new java.lang.String(delivery.getBody(),"UTF-8"));}}, new CancelCallback() {public void handle(java.lang.String s) throws IOException {System.out.println("接受消息失败了...");}});}catch (Exception ex){ex.printStackTrace();}finally {// 5: 释放连接关闭通道if (channel != null && channel.isOpen()) {try {channel.close();} catch (Exception ex) {ex.printStackTrace();}}if (connection != null) {try {connection.close();} catch (Exception ex) {ex.printStackTrace();}}}}
}
4. 运行
运行消息生产者的代码之后，RabbitMQ中有1条的ready的消息。当运行消费者的代码后，ready的消息就被消费了。
5. 持久化队列和非持久化队列
持久化队列和非持久化队列的区别是，持久化队列会被保存在磁盘中，固定并持久的存储，当Rabbit服务重启后，该队列会保持原来的状态在RabbitMQ中被管理，而非持久化队列不会被保存在磁盘中，Rabbit服务重启后队列就会消失。
- 非持久化比持久化的优势就是，由于非持久化不需要保存在磁盘中，所以使用速度就比持久化队列快。即是非持久化的性能要高于持久化。
- 而持久化的优点就是会一直存在，不会随服务的重启或服务器的宕机而消失。
