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

网站的建设方法包括什么作用,网站的建设周期与阶段,iphone开发网站,网址你懂我意思正能量一、MQ与RabbitMQ概述 1. MQ简述 MQ#xff08;Message Queue#xff09;消息队列#xff0c;是基础数据结构中 “先进先出” 的一种数据结构#xff0c;也是在消息的传输过程中保存消息的容器#xff08;中间件#xff09;#xff0c;多用于分布式系统之间进行通信。 …一、MQ与RabbitMQ概述 1. MQ简述 MQMessage Queue消息队列是基础数据结构中 “先进先出” 的一种数据结构也是在消息的传输过程中保存消息的容器中间件多用于分布式系统之间进行通信。 一般MQ用来解决系统耦合、异步消息、流量削峰等问题实现高性能、高可用、可伸缩和最终一致性架构。AP架构 分布式系统有两种通信方式直接远程调用 和 借助第三方MQ完成间接通信。发送方称为生产者接收方称为消费者 2. MQ的优势与劣势 2.1 MQ的优势 MQ的优势(应用解耦、异步、削峰) 应用解耦提高系统容错性和可维护性异步提速提升用户体验和系统吞吐量削峰填谷提高系统稳定性。 1、应用解耦 2、异步提速 3、削峰填谷 填谷 使用了MQ之后限制消费消息的速度为1000这样一来高峰期产生的数据势必会被积压在MQ中高峰就被“削”掉了但是因为消息积压在高峰期过后的一段时间内消费消息的速度还是会维持在1000直到消费完积压的消息这就叫做“填谷”从而提升系统的稳定性。 2.2 MQ的劣势 引入MQ会遇到下列问题 消息可靠性问题如何确保发送的消息至少被消费者消费一次避免消息丢失问题延迟消息问题 如何实现消息的延迟投递解决方案使用延时队列、TTL、延迟队列插件实现高可用问题如何避免单点MQ故障而导致的不可用问题解决方案搭建MQ集群消息堆积问题如何解决数百万消息堆积无法及时消费的问题 3. 常见的MQ产品 市面上有很多MQ产品例如RabbitMQ、RocketMQ、ActiveMQ、Kafka、ZeroMQ、MetaMq等也有直接使用Rdis充当消息队列的场景。在实际技术选型时需要结合自身需求及MQ产品特征来综合考虑。 几种常见MQ的对比 RabbitMQActiveMQRocketMQKafka公司/社区RabbitApache阿里Apache开发语言Erlang二郎神高并发语言JavaJavaScalaJava协议支持AMQPXMPPSMTPSTOMPOpenWireSTOMPRESTXMPPAMQP自定义协议自定义协议社区封装了http协议支持可用性高一般高高单机吞吐量一般万级差高十万级非常高十万级消息延迟微秒级毫秒级毫秒级毫秒以内消息可靠性高一般高一般功能特性并发能力强性能极其好延迟低社区活跃管理界面丰富老牌产品成熟度高文档较多MQ功能比较完备扩展性佳只支持主要的MQ功能毕竟是为大数据领域准备的。 追求可用性(高-低)Kafka、 RocketMQ 、RabbitMQ 追求可靠性RabbitMQ、RocketMQ 追求吞吐能力RocketMQ、Kafka 追求消息低延迟RabbitMQ、Kafka。 4. RabbitMQ简述 RabbitMQ官网地址http://www.rabbitmq.com/ RabbitMQ是基于AMQP协议使用Erlang语言开发的一款消息队列产品。 AMQP (全称Advanced Message Queuing Protocol表示高级消息队列协议是一个网络协议是应用层协议的一个开放标准为面向消息的中间件设计。基于此协议的客户端与消息中间件可传递消息并不受客户端/中间件不同产品不同的开发语言等条件的限制。2006年AMQP规范发布。类比HTTP。同类的协议还有MQTT用于物联网场景下 RabbitMQ中的一些角色AMQP协议消息中间件类似 publisher生产者consumer消费者exchange 交换机负责消息路由queue队列存储消息virtualHost虚拟主机隔离不同租户的exchange、queue、消息的隔离。 RabbitMQ工作模式 RabbitMQ提供了6种工作模型但是我们常用的只有5种简单队列模型、工作队列模型、发布订阅模型广播、路由、主题。第六种RPC远程调用不属于mq 官网对应模式介绍https://www.rabbitmq.com/getstarted.html JMSJavaMessage Service JMSJava消息服务应用程序接口即Java操作消息中间件的APIJMS是JavaEE规范的一种类比JDBC很多消息中间件都实现了JMS规范例如ActiveMQ。RabbitMQ官方没有提供JMS的实现包但是开源社区有。 二、RabbitMQ安装与配置 1. 基于docker快速安装RabbitMQ 扩展docker-compose安装rabbitmqhttps://gitee.com/aopmin/docker-compose/blob/master/Linux/RabbitMQ/docker-compose.yml 1、拉取镜像 docker pull rabbitmq:3.8-management2、运行容器 docker run -e RABBITMQ_DEFAULT_USERadmin -e RABBITMQ_DEFAULT_PASS123456 -v /docker/rabbitmq/plugins:/plugins --name rabbitmq --hostname my-rabbit -p 15672:15672 -p 5672:5672 -d \rabbitmq:3.8-management\ 代表换行-e 指定环境变量-e RABBITMQ_DEFAULT_USERadmin用户名-e RABBITMQ_DEFAULT_PASS123456密码-v 挂载目录或文件 (数据卷)-p 15672:15672 用于web管理页面使用的端口 (管理员页面)-p 5672:5672 用于生产和消费端使用的端口通信端口也就是在代码里使用-d 后台运行–name rabbitmq 容器名字–hostname my-rabbitRabbitMQ的一个重要注意事项是它根据所谓的 “节点名称” 存储数据默认为主机名 3、启动xxx插件

进入容器

docker exec -it rabbitmq /bin/bash# 启动xxx插件 rabbitmq-plugins enable xxxRabbitMQ管理端 管理端访问地址http://ip:15672/ 2. 创建用户和虚拟机 1、添加一个新用户 添加成功后列表会显示该用户但是这个用户没有操作权限需要为他创建一个虚拟机 2、创建虚拟机 为指定用户授权 最后该用户就可以操作这个虚拟机了 三、RabbitMQ快速入门 使用简单模型中的基本模式完成消息传递 官方的HelloWorld示例是基于简单消息队列模型来实现的其中包括三个角色 publisher消息发布者将消息发送到队列queuequeue消息队列负责接受并缓存消息consumer订阅队列处理队列中的消息。 1. 基础环境搭建 1、创建父工程mq-demo并在pom文件中导入如下依赖 parentgroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-parent/artifactIdversion2.3.9.RELEASE/versionrelativePath/ /parentdependenciesdependencygroupIdorg.projectlombok/groupIdartifactIdlombok/artifactId/dependency!–SpringAMQP依赖包含RabbitMQ–dependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-amqp/artifactId/dependency!–单元测试–dependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-test/artifactId/dependency /dependencies2、创建子模块publisher、consumer并编写启动类和yml配置文件

日志输出格式配置

logging:pattern:dateformat: MM-dd HH:mm:ss:SSS2. publisher消息发布者实现 消息收发流程Connection连接、Channel通道、queue队列、exchange 交换机。 publisher消息发布者实现思路 建立连接创建Channel声明队列发送消息关闭连接和channel 1、编写publisher测试代码 package com.baidou.mq.test;import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConnectionFactory; import org.junit.Test;import java.io.IOException; import java.util.concurrent.TimeoutException;/*** 生产者* author 白豆五* version 2023/04/27* since JDK8/ public class PublisherTest {Testpublic void testSendMessage() throws IOException, TimeoutException {// 1.创建连接工厂ConnectionFactory factory new ConnectionFactory();// 1.1.设置连接参数分别是主机名、端口号、vhost、用户名、密码factory.setHost(192.168.200.128);factory.setPort(5672);factory.setVirtualHost(/);factory.setUsername(admin);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();} }2、在建立连接处打断点并以debug方式启动方便观察每个组件的创建 查看连接信息 继续按F8查看通道信息 继续按F8查看队列信息 最后直接放行程序查看队列中的消息 3. consumer消费者实现 consumer消费者实现思路 建立连接创建Channel声明队列订阅消息 1、编写消费者代码 package com.baidou.mq.test;import com.rabbitmq.client.;import java.io.IOException; import java.util.concurrent.TimeoutException;/*** 消费者* author 白豆五* version 2023/04/27* since JDK8*/ public class ConsumerTest {public static void main(String[] args) throws IOException, TimeoutException {// 1.创建连接工厂ConnectionFactory factory new ConnectionFactory();// 1.1.设置连接参数分别是主机名、端口号、vhost、用户名、密码factory.setHost(192.168.200.128);factory.setPort(5672);factory.setVirtualHost(/);factory.setUsername(admin);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.订阅消息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 】);}});System.out.println(等待接收消息。。。。);} }2、测试启动程序后会一直执行不用的时候将程序结束即可 四、SpringAMQP与RabbitMQ工作模型 1. SpringAMQP概述 AMQP是消息中间件收发消息的协议(规范)具体实现由各个消息中间厂商实现例如 RabbitMQ SpringAMQP是基于RabbitMQ封装的一套模板并且还利用SpringBoot对其实现了自动装配使用起来非常方便。 SpringAMQP的官方地址https://spring.io/projects/spring-amqp SpringAMQP提供了三个功能 自动声明队列、交换机及其绑定关系基于注解的监听器模式异步接收消息封装了RabbitTemplate工具用于发送消息 。 RabbitMQ工作模型简单队列模型、工作队列模型、发布订阅模型广播、路由、主题。 2. BasicQueue 简单队列模型 使用SpringAMQP实现基础消息队列功能 1、在父工程中引入spring-amqp起步依赖 !–SpringAMQP依赖包含RabbitMQ– dependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-amqp/artifactId /dependency2、消息发送 2.1、在publisher服务的application.yml中添加rabbitmq配置

配置rabbitmq

spring:rabbitmq:host: 192.168.200.128 # 主机名port: 5672 # 端口virtual-host: / # 虚拟主机username: admin # 用户名password: 123456 # 密码# 配置日志格式 logging:pattern:dateformat: MM-dd HH:mm:ss:SSS2.2、在publisher服务中编写测试类SpringAmqpTest并利用RabbitTemplate实现消息发送 package com.baidou.mq.test;import org.junit.Test; import org.junit.runner.RunWith; import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.test.context.SpringBootTest; import org.springframework.test.context.junit4.SpringRunner;/*** 使用SpringAMQP实现简单队列模型的消息发送** author 白豆五* version 2023/04/27* since JDK8/ RunWith(SpringRunner.class) SpringBootTest public class SpringAmqpTest {// 操作RabbitMQ的模板类Autowiredprivate RabbitTemplate rabbitTemplate;/** 测试简单队列模型*/Testpublic void testSimpleQueue() {// 队列名称String queueName simple.queue;// 消息String message hello, spring amqp!;// 发送消息rabbitTemplate.convertAndSend(queueName, message);} }3、消息接收 3.1、在consumer服务的application.yml中添加rabbitmq配置

配置rabbitmq

spring:rabbitmq:host: 192.168.200.128 # 主机名port: 5672 # 端口virtual-host: / # 虚拟主机username: admin # 用户名password: 123456 # 密码# 配置日志格式 logging:pattern:dateformat: MM-dd HH:mm:ss:SSS3.2、在consumer服务的com.baidou.mq.listener包中创建SpringRabbitListener类 package com.baidou.mq.listener;import org.springframework.amqp.rabbit.annotation.RabbitListener; import org.springframework.stereotype.Component;/*** 消息监听类** author 白豆五* version 2023/04/27* since JDK8/ Component public class SpringRabbitListener {/** 订阅消息** param msg* throws InterruptedException*/RabbitListener(queues simple.queue) // 指定监听的队列名称为simple.queuepublic void listenSimpleQueueMessage(String msg) throws InterruptedException {System.out.println(spring 消费者接收到消息【 msg 】);} }4、测试 先启动consumer服务(启动类)然后在publisher服务中运行测试代码发送MQ消息。 3. WorkQueue 工作队列模型