手机网站建设多钱百度直播
一、什么是RabbitMQ
RabbitMQ 是一款基于 AMQP(高级,消息队列协议) 的开源消息中间件,专为分布式系统设计,用于实现应用程序间的异步通信,其核心功能是通过 消息代理(Message Broker) 机制,实现可靠的消息传递、存储和路由,支持跨语言和跨平台交互。
二、RabbitMQ的工作流程
1. Producer 和 Broker 建立一个连接(Connection)并声明一个信道(channel);
2. Producer 声明 一个交换机(Exchange);
3. Producer 声明一个队列(Queue);
4. Producer 与 Broker 连接建立完毕,开始向 Broker 传输信息。
三、RabbitMQ使用方法(入门)
3.1 引入依赖
创建一个maven项目并添加RabbitMQ依赖:
<dependency><groupId>com.rabbitmq</groupId><artifactId>amqp-client</artifactId><version>5.20.0</version></dependency>
3.2 创建实体类ProducerDemo,ConsumerDemo
package rabbitmq;import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class ProducerDemo {public static void main(String[] args) throws IOException, TimeoutException {/** 根据rabbitmq的工作流程图,主要工作分为4步:* 1.服务器ip* 2.服务器端口号* 3.账号、密码* 4.虚拟主机*///1.建立连接ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost("110.41.17.130");connectionFactory.setPort(5672);connectionFactory.setPassword("study");connectionFactory.setUsername("study");connectionFactory.setVirtualHost("java113");Connection connection = connectionFactory.newConnection();//2.开启信道Channel channel = connection.createChannel();//3.声明交换机(使用内置交换机)//4.声明队列/*AMQP.Queue.DeclareOk queueDeclare(String var1, boolean var2, boolean var3, boolean var4, Map<String, Object> var5) throws IOException;**参数说明:* s: 队列名* b: 可持久化* b1: 是否独占( 是否只能有一个消费者读取队列中的信息)* b2: 没有消费者时是否自动删除* map: 参数*/channel.queueDeclare("hello",true,false,false,null);//5.发送消息/** void basicPublish(String var1, String var2, AMQP.BasicProperties var3, byte[] var4) throws IOException;* 参数说明:* var1 交换机名称* var2 内置交换机,和队列名称一致(路由规则)* var3 属性配置* var4 消息*/for (int i = 0; i < 10; i++) {String msg = "hello rabbitmq " + i;channel.basicPublish("","hello",null,msg.getBytes());}System.out.println("消息发送成功");//6.进行资源释放channel.close();connection.close();}
}
package rabbitmq;import com.rabbitmq.client.*;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class ConsumerDemo {public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {//1.创建连接ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost("110.41.17.130");connectionFactory.setPort(5672);connectionFactory.setUsername("study");connectionFactory.setPassword("study");connectionFactory.setVirtualHost("java113");Connection connection = connectionFactory.newConnection();//2.创建channelChannel channel = connection.createChannel();//3.声明一个队列(如果生产者声明了可以省略)channel.queueDeclare("hello",true,false,false,null);//4.消费消息/** 参数说明:* 1. 队列名称* 2. 是否自动确认(即确认consumer是否收到消息)* 3. 接收到消息后,执行的逻辑*/DefaultConsumer consumer = new DefaultConsumer(channel){@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {//TODOSystem.out.println("接收消息" + new String(body));}};channel.basicConsume("hello",true,consumer);Thread.sleep(1000);//5.释放资源channel.close();connection.close();}
}
1> 运行生产者代码:
查看RabbitMQ管理界面的队列信息:
可以看到,队列中新增了10条信息
2>运行消费者代码
查看队列信息:
可以看到,队列中的信息已经被消费完毕
四、总结
在RabbitMQ的使用中,主要有以下几步:
1> 建立连接:无论是生产者还是消费者,要与服务器建立连接,需要知道:
(1)服务器的 IP 地址和 端口号;
(2)用户名 和 密码;
(3)要连接到服务器的哪台虚拟机。
2>创建信道(channel)
3>声明交换机(使用内置交换机可不用声明)
4>声明队列(如果生产者已经声明队列,消费者可以不声明(不建议))
5>生产者使用 channel.basicPublish 方法发送信息,消费者使用chaanel.basicConsumer 方法消费信息