考试类网站如何做,世界500强企业国家排名,国外网站建设素材,如何成为一个优秀的网站设计师RabbitMQ RabbitMQRabbitMQ安装 常见的消息模型基本消息队列SpringAMQPWorkQueue消息预取发布订阅模式Fanout ExchangeDirectExchangeTopicExchange 消息转换器 RabbitMQ
RabbitMQ是基于Erlang语言开发的开源消息通信中间件 官网地址#xff1a;https://www.rabbitmq.com/
R… RabbitMQ RabbitMQRabbitMQ安装 常见的消息模型基本消息队列SpringAMQPWorkQueue消息预取发布订阅模式Fanout ExchangeDirectExchangeTopicExchange 消息转换器 RabbitMQ
RabbitMQ是基于Erlang语言开发的开源消息通信中间件 官网地址https://www.rabbitmq.com/
RabbitMQ安装
我们在Centos虚拟机中使用Docker来安装
下载镜像在线拉取 docker pull rabbitmq安装MQ
docker run\
--env RABBITMQ_DEFAULT_USERitcast \ # 设置环境变量用户名
--env RABBITMQ_DEFAULT_PASS \ # 设置环境变量密码
--name mq \ # 队列名称
--hostname mq1 \ #配置主机名
-p 15672:15672 \ # MQ管理端口
-p 5672:5672 \ #MQ消息传输端口
-d \ # 后台运行
rabbitmq交换机的创建与消息的发送由虚拟主机来完成每个用户的虚拟主机是相互隔离的 在RabbitMQ中 channel操作MQ的工具 exchange路由消息到队列中 queue缓存消息 virtual host:虚拟主机是对queueexchange等资源的逻辑分组
常见的消息模型
基本消息队列工作消息队列 这两种并没有用到交换机而是直接到达队列 发布订阅PublishSubscribe根据交换机类型不同分为三种 Fanout Exchange:广播 Direct Exchange:路由 Topic Exchange:主题
基本消息队列
publisher:消息发布者将消息发送到队列queue queue消息队列负责接收并缓存消息 consumer订阅队列处理队列中的消息
java模型(消息发布者)
Test
public void test() throws IOException,TimeoutException{//1.建立连接,与消息队列进行连接ConnetionFactory factory new ConnetionFactory();//设置连接参数主机名端口号vhost用户名密码factory.setHost(192.168.75.136);factory.setPort(5672);factory.setVirtualHost(/);factory.setUsername(itcast);factory.setPassword();//建立连接Connection connection factory.newConnection();//创建通道Channel,就可以向队列发送消息了Channel channel connection.createChannel();//创建队列String queuenamehlh;channel.queueDeclare(queuename,false,false,false,null);//发送消息String messagehello;channel.basicPublish(,queuename,null,message.getBytes());//关闭通道和连接channel.close();connection.close();
}java模型(消息消费者) //1.建立连接,与消息队列进行连接ConnetionFactory factory new ConnetionFactory();//设置连接参数主机名端口号vhost用户名密码factory.setHost(192.168.75.136);factory.setPort(5672);factory.setVirtualHost(/);factory.setUsername(itcast);factory.setPassword();//建立连接Connection connection factory.newConnection();//创建通道Channel,就可以向队列发送消息了Channel channel connection.createChannel();//创建队列String queuenamehlh;channel.queueDeclare(queuename,false,false,false,null);//订阅消息channel.basicConsume(queuename,true,new DefaultConsumer(channel){Override//处理消息的代码,绑定函数有了消息才执行public void handleDelivery(String consumerTag,Envelope envelope,AMQP.BasicProperties properties,byte[] body)throws IOException{//处理消息String messagenew String(body); }})注意上边生产者消费者都创建了队列 这是为了防止消息队列中的队列不存在在进行消息队列初始化的时候不知道是先建立消费者还是先建立生产者所以都执行创建函数但是创建的队列只有一个不会重复 SpringAMQP
AMQP 是用于在应用程序或之间传递业务消息的开放标准该协议与语言和平台无关更符合微服务中的独立性的要求 Spring AMQP Spring AMQP是基于AMQP协议定义的一套API规范提供了模板来发送和接收消息包含两部分其中Spring-amqp是基础抽象spring-rabbit是底层的默认实现 特征
监听器容器用于异步处理入站消息用于发送和接收消息的RabbitTemplateRabbitadmin用于自动声明队列交换和绑定
使用
引入spring-amqp的依赖 在yml中配置mq连接信息
spring: rabbitmq:host: 192.168.75.136 #主机名port: 5672 #端口virtual-host: / #虚拟主机username: itcast #用户名password: #密码在生产者服务中利用RabbitTemplate发送消息到hlh.queue这个队列
public class springamqptest{Autowiredprivate RabbitTemplate rabbittemplate;Testpublic void test(){String queuenamehlh.queue;String messagehello;rabbittemplate.convertAndSend(queuename,message);}
}在消费者服务端编写消费逻辑绑定到hlh.queue这个队列中
Component
public class SpringrabbitListener {RabbitListener(queueshlh.queue)public void listenSimple(String msg) throws InterruptedException{//消费逻辑代码}
}注意消息一旦消费就会从队列中删除rabbitmq没有消息回溯功能
WorkQueue
Work queue工作队列。可以提高消息处理速度避免队列消息堆积 一个消息队列绑定多个消费者 假设现在生产者每秒循环发送50条消息此时的消费者怎么处理
Component
public class SpringrabbitListener {RabbitListener(queueshlh.queue)public void listenSimple(String msg) throws InterruptedException{//消费逻辑代码}RabbitListener(queueshlh.queue)public void listenSimple2(String msg) throws InterruptedException{//消费逻辑代码}
}通过定义多个消费者进行消费追上生产者生产的速度同一个消息只能被一个消费者消费一旦消费完就会在队列中删除
消息预取
指的每个消费者每次取多少条消息 可以通过配置进行配置
spring:rabbitmq:host: 192.168.75.136port: 5672virtual-host: /username: itcastpassword: listener:simple:prefecth: 1 #每次只能获取一条消息处理完才能获得下一个消息发布订阅模式
发布订阅可以使得同一个消息发送给多个消费者实现方式是加入了exchange交换机
注意exchange负责消息路由而不是存储路由失败则消息丢失
交换机的作用
接收生产者的消息将消息按照规则路由到与之绑定的队列不能缓存消息路由失败消息丢失FanoutExchange的会将消息路由到每个绑定的队列
SpringAMQP提过了声明交换机队列绑定关系的API
Fanout Exchange
Fanout Exchange 会将所有的消息路由到每一个跟其绑定的queue 在创建配置类在配置类中进行消息队列绑定交换机
Configuration
public class FanoutConfig{// 声明FanoutExchange交换机Beanpublic FanoutExchange fanoutExchange(){return new FanoutExchange(itcast.fanout);}//声明一个队列Beanpublic Queue fanoutQueue1(){return new Queue(fanout.queue1);}// 绑定队列跟交换机Beanpublic Binding bindingQueue1(Queue fanoutQueue1,FanoutExchange fanoutExchange){return BindingBuilder.bind(fanoutQueue1).to(fanoutExchange);}
}此时的生产者如何发送消息
public void test(){//给出交换机名称String exchangeNameitcast.fanout;String messagehello;//发送消息rabbitTemplate.convertAndSend(exchangeName,,message);
}监听者如何收到消息
RabbitListener(queuesfanout.queue1)
public void listener(String msg){//处理得到的消息
}DirectExchange
Direct Exchange 会将接收到的消息根据规则路由到指定的Queue因此成为路由模式routes 每一个Queue都与Exchange设置一个BindingKey 发布者发送消息时指定消息的RoutingKeyExchange将消息路由到BindingKey与消息RoutingKey一致的队列 一个队列可以指定多个Key 我们可以通过 RabbitListener声明ExchangeQueueRoutingKey 在消费者方法上注解
RabbitListener(bindingsQueueBinding(valueQueue(namedirect.queue1),exchangeExchange(nameitcast.direct,typeExchangeTypes.DIRECT),key{red,blue}))
public void Listener(String msg){//进行消息的处理
}在生产者生产时
public void test(){//给出交换机名称String exchangeNameitcast.fanout;String messagehello;//发送消息rabbitTemplate.convertAndSend(exchangeName,blue,message);
}TopicExchange
TopicExchange与路由模式类似区别在于routingKey必须是多个单词的列表并且以.分隔 Queue与Exchange指定BindingKey时可以使用通配符 #:代指0个或多个单词 *:代指一个单词 同样也是使用 RabbitListener进行声明
RabbitListener(bindingsQueueBinding(valueQueue(namedirect.queue1),exchangeExchange(nameitcast.direct,typeExchangeTypes.DIRECT),keyhi.#))
public void Listener(String msg){//进行消息的处理
}生产者生产消息
public void test(){//给出交换机名称String exchangeNameitcast.fanout;String messagehello;//发送消息rabbitTemplate.convertAndSend(exchangeName,hi.now,message);
}消息转换器
在SpringAMQP的发送方法中接收消息的类型是Object也就是我们可以发送任意对象类型的消息SpringAMQP会帮助我们序列化为字节后发送 Spring的对消息对象的处理是由org.springframework.amqp.support.converter.MessageConverter来处理的而默认实现是SimpleMessageConverter,基于JDK的ObjectOutputStream完成序列化 如果要修改只需定义一个MessageConverter 类型的Bean即可推荐使用JSON方式完成序列化
引入jackson的依赖 声明MessageConverter
Bean
public MessageConverter jsonMessageConverter(){return new Jackson2JsonMessageConverter();
}这样发送的消息就会使用自定义的转换类型
