当前位置：首页 > news >正文

RabbitMQ入门

news 2025/6/23 22:10:47

1. 搭建示例工程

1.1. 创建工程

创建项目：rabbitmq-producer
创建项目：rabbitmq-consumer

1.2. 添加依赖

往两个rabbitmq的pom.xml文件中添加如下依赖：

    <dependencies><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>

2. 编写生产者

编写消息生产者 com.donglin.rabbitmq.simple.Producer

package com.donglin.rabbitmq.simple;import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;public class Producer {public static void main(String[] args) throws Exception {//创建连接工厂ConnectionFactory connectionFactory = new ConnectionFactory();//主机地址connectionFactory.setHost("192.168.121.140");//连接端口;默认为 5672connectionFactory.setPort(5672);//虚拟主机名称;默认为 /connectionFactory.setVirtualHost("/");//连接用户名；默认为guestconnectionFactory.setUsername("admin");//连接密码；默认为guestconnectionFactory.setPassword("admin");//创建连接Connection connection = connectionFactory.newConnection();//创建频道Channel channel = connection.createChannel();// 声明（创建）队列/*** queue      参数1：队列名称* durable    参数2：是否定义持久化队列,当mq重启之后,还在* exclusive  参数3：是否独占本次连接*            ① 是否独占,只能有一个消费者监听这个队列*            ② 当connection关闭时,是否删除队列* autoDelete 参数4：是否在不使用的时候自动删除队列,当没有consumer时,自动删除* arguments  参数5：队列其它参数*/channel.queueDeclare("simple_queue", true, false, false, null);// 要发送的信息String message = "你好；小兔子！";/*** 参数1：交换机名称,如果没有指定则使用默认Default Exchage* 参数2：路由key,简单模式可以传递队列名称* 参数3：配置信息* 参数4：消息内容*/channel.basicPublish("", "simple_queue", null, message.getBytes());System.out.println("已发送消息：" + message);// 关闭资源channel.close();connection.close();}
}

运行程序：http://192.168.6.100:15672
在执行上述的消息发送之后；可以登录rabbitMQ的管理控制台,可以发现队列和其消息：
在这里插入图片描述

3. 编写消费者

编写消息的消费者 com.donglin.rabbitmq.simple.Consumer

package com.donglin.rabbitmq.simple;import com.rabbitmq.client.*;import java.io.IOException;public class Consumer {public static void main(String[] args) throws Exception {//1.创建连接工厂ConnectionFactory factory = new ConnectionFactory();//2. 设置参数factory.setHost("192.168.121.140");//ipfactory.setPort(5672); //端口  默认值 5672factory.setVirtualHost("/");//虚拟机 默认值/factory.setUsername("admin");//用户名factory.setPassword("admin");//密码//3. 创建连接 ConnectionConnection connection = factory.newConnection();//4. 创建ChannelChannel channel = connection.createChannel();//5. 创建队列Queue/*queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments)参数：1. queue：队列名称2. durable：是否持久化,当mq重启之后,还在3. exclusive：* 是否独占。只能有一个消费者监听这队列* 当Connection关闭时,是否删除队列4. autoDelete：是否自动删除。当没有Consumer时,自动删除掉5. arguments：参数。*///如果没有一个名字叫simple_queue的队列,则会创建该队列,如果有则不会创建channel.queueDeclare("simple_queue",true,false,false,null);// 接收消息DefaultConsumer consumer = new DefaultConsumer(channel){/*回调方法,当收到消息后,会自动执行该方法1. consumerTag：标识2. envelope：获取一些信息,交换机,路由key...3. properties：配置信息4. body：数据*/@Overridepublic 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));}};/*basicConsume(String queue, boolean autoAck, Consumer callback)参数：1. queue：队列名称2. autoAck：是否自动确认 ,类似咱们发短信,发送成功会收到一个确认消息3. callback：回调对象*/// 消费者类似一个监听程序,主要是用来监听消息channel.basicConsume("simple_queue",true,consumer);}
}