千博企业网站管理系统2013,网络营销方案包括哪些内容,网页模板布局,苏州集团网站制作RabbitMQ
RabbitMQ是基于Erlang语言开发的开源消息通信中间件,有几个常见概念 connections(连接): 将来publisher(消息的发送者)或者consumer(消息的接收者)都需要先与MQ建立连接 channel(通道): 建立连接后需要创建通道,生产者和消费者就是基于通道完成消息的发送和接收 ex…RabbitMQ
RabbitMQ是基于Erlang语言开发的开源消息通信中间件,有几个常见概念 connections(连接): 将来publisher(消息的发送者)或者consumer(消息的接收者)都需要先与MQ建立连接 channel(通道): 建立连接后需要创建通道,生产者和消费者就是基于通道完成消息的发送和接收 exchange(交换机): 路由消息到队列中 queue(队列)缓存消息(二进制数据块) virtual host(虚拟主机): 对queue,exchange等资源逻辑分组隔离防止不同用户操作MQ时发生冲突,一般每个用户都有自己独享的虚拟主机(默认只有/) 5种消息模型
RabbitMQ官方提供了5个不同的Demo实例,对应了不同的消息模型
基于队列完成消息的发送和接收: 基本消息类型(BasicQueue), 工作消息队列(WorkQueue)发布订阅模型(Publish,Subscribe): 基于交换机的类型不同分为广播(Fanout Exchange),路由(Direct Exchange),主题(Topic Exchange)
需求: 基于最基础的消息队列模型实现HelloWorld案例
publisher(消息发布者): 将消息发送到队列queuequeue(消息队列): 负责接受并缓存消息consumer(消息订阅者): 处理队列中的消息 第一步: 创建父工程mq-demo用来管理项目依赖,创建子模块publisher(消息的发送者)和consumer(消息的消费者)
第二步: 在publisher模块的测试类中编写消息发送流程的逻辑代码
建立connection–创建channe–使用channel声明队列–使用channel向声明的队列发送消息(底层需要把消息转成二进制传输)
public class PublisherTest {Testpublic void testSendMessage() throws IOException, TimeoutException {// 1.建立连接ConnectionFactory factory new ConnectionFactory();// 1.1.设置连接MQ的参数主机名、消息通信端口号、vhost、用户名、密码factory.setHost(192.168.150.101);factory.setPort(5672);factory.setVirtualHost(/);factory.setUsername(root);factory.setPassword(123456);// 1.2.建立连接Connection connection factory.newConnection();// 2.创建通道ChannelChannel channel connection.createChannel();// 3.声明队列String queueName simple.queue;channel.queueDeclare(queueName, false, false, false, null);// 4.向声明的队列发送消息(底层是用字节传输)String message hello, rabbitmq!;channel.basicPublish(, queueName, null, message.getBytes());System.out.println(发送消息成功【 message 】);// 5.关闭通道和连接channel.close();connection.close();}
}第三步: 在consumer模块的测试类中编写消息接收流程的业务逻辑代码
建立connection–创建channel-使用channel声明队列–订阅信息(定义消费行为和绑定队列)—删除队列中的消息(阅后即焚)订阅信息流程: 定义consumer的消费行为handleDelivery()即接收到消息后要执行的回调函数–利用channel将消费者与队列绑定因为消费者和生产者启动的顺序不确定,为了避免寻找的队列可能不存在,所以消费者和生产者都需要各自声明创建队列(无论谁先声明实际只会创建一个队列)
public class ConsumerTest {public static void main(String[] args) throws IOException, TimeoutException {// 1.建立连接ConnectionFactory factory new ConnectionFactory();// 1.1.设置连接参数分别是主机名、端口号、vhost、用户名、密码factory.setHost(192.168.150.101);factory.setPort(5672);factory.setVirtualHost(/);factory.setUsername(root);factory.setPassword(root);// 1.2.建立连接Connection connection factory.newConnection();// 2.创建通道ChannelChannel channel connection.createChannel();// 3.声明队列,只有当声明的队列不存在时才会被创建我们在这里再次声明队列可以防止队列为nullString queueName simple.queue;channel.queueDeclare(queueName, false, false, false, null);// 4.订阅消息(定义消费行为和绑定队列)channel.basicConsume(queueName, true, new DefaultConsumer(channel){Overridepublic void handleDelivery(String consumerTag, Envelope envelope,AMQP.BasicProperties properties, byte[] body) throws IOException {// 5.处理消息(将字节转化为字符串)String message new String(body);System.out.println(接收到消息【 message 】);}});// 异步执行,只有RabbitMQ把消息发送过来后以上定义的回调函数才会执行System.out.println(等待接收消息....);}
}
