pc手机一体网站南宁两学一做网站

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

pc手机一体网站,南宁两学一做网站,临沂网站服务器价格,京东短网址在线生成市面常见消息队列中间件对比 技术名称吞吐量 /IO/并发时效性#xff08;类似延迟#xff09;消息到达时间可用性可靠性优势应用场景activemq万级高高高简单易学中小型企业、项目rabbitmq万级极高#xff08;微秒#xff09;高极高生态好#xff08;基本什么语言都支持类似延迟消息到达时间可用性可靠性优势应用场景activemq万级高高高简单易学中小型企业、项目rabbitmq万级极高微秒高极高生态好基本什么语言都支持、时效性高、易学适合绝大数的分布式应用kafka10万 QBS高毫秒极高极高吞吐量大、可靠性、可用性、强大的数据流处理能力适合大规模处理数据的场景、比如构建日志手机系统、实时数据传输、事件流收集传输rocketmq10万 QBS高ms极高极高吞吐量大、可靠性、可用性、可扩展性适用于金融等可靠性要求较高的场景、适合大规模的消息处理。金融、电商、大规模社交pulsar10万 QBS高ms高极高可靠性、可用性很高、新兴技术架构先进适合大规模、高并发的分布式系统云原生适合实时分析、事件流处理、物联网数据处理。 RabbitMQ  RabbitMQ 是基于 AMQP 高级消息队列协议的。 实际使用可根据官方文档的 demo 。 官方文档RabbitMQ Tutorials | RabbitMQ 模型 生产者通俗就是发消息的人比如在外卖软件上点餐的人 消费者通俗就是处理消息的任务比如外卖软件上的商家需要根据顾客的要求制作餐 交换机负责把消息转发到对应的队列中比如有外卖的时候系统给附近的外面小哥派单 队列存放消息的地方等待消费者消费比如商家肯定不是只做一份餐做好的餐放在一个指定的位置等待外卖小哥来取餐 路由转发就是怎么把消息从一个地方转到另一个地方通常加在交换机和队列之间比如系统指定某个范围的外卖小哥接这单 安装

1. 首先安装 RabbitMQ直接官网下载即可如果下载速度慢可以换个网络或者找找有没有国内镜像。当初我下载的时候找了半天的镜像都是版本比较老的结果想着挂一晚上下载结果官网突然就快了白折腾了。 官方网站Installing on Windows | RabbitMQ 国内镜像Index of rabbitmq-server-local/v3.12.7 一路 next 傻瓜式安装即可 安装之后检查服务中是否已经运行了。 2. 安装监控面板 在 RabbitMQ 安装目录下的 sbin 目录下的CMD 输入下面的命令 rabbitmq-plugins.bat enable rabbitmq_management 安装成功 默认端口号是 5672webUI 是 15672 在浏览器输入地址打开管理界面http://localhost:15672 默认账号密码是 guest 注意1. 安装目录不能是中文不能有空格等非法字符否则页面打不开 2. 如果想要在远程服务器访问 RabbitMQ 管理面板需要创建管理员账号比如在宝塔面板使用时宝塔面板提供的 admin账号地址就是宝塔面板的 IP  创建账号access-control | RabbitMQ 入门 依赖引入 !– https://mvnrepository.com/artifact/com.rabbitmq/amqp-client – dependencygroupIdcom.rabbitmq/groupIdartifactIdamqp-client/artifactIdversion5.20.0/version /dependency 单消费者和生产者 一对一的关系
2. 生产者代码 public class SingleProducer {private final static String QUEUE_NAME hello;public static void main(String[] argv) throws Exception {//创建连接ConnectionFactory factory new ConnectionFactory();//设置了本地连接如果修改了用户名和密码需要设置/factory.setPassword();factory.setUsername();/factory.setHost(localhost);//建立连接、创建频道//频道类似客户端用于调用serverConnection connection factory.newConnection();Channel channel connection.createChannel();//创建队列队列持久化第二份参数设置为 truechannel.queueDeclare(QUEUE_NAME, false, false, false, null);String message Hello World!;//发送消息channel.basicPublish(,QUEUE_NAME,null,message.getBytes(StandardCharsets.UTF_8));System.out.println( [] Waiting for messages. To exit press CTRLC);} } channel 频道理解为操作消息队列的 Client通过 channel 收发消息提供了和消息对了 server 建立通信的传输方法 channel.queueDeclare 方法参数 queue这是一个字符串参数代表要声明的队列的名称。如果队列不存在则会自动创建一个新的队列。 durable这是一个布尔值参数表示队列是否持久化。如果设置为true则队列会在服务器重启后仍然存在如果设置为false则队列在服务器重启后会被删除。默认值为false。 exclusive这也是一个布尔值参数表示队列是否为独占模式。如果设置为true则只有当前连接可以访问该队列如果设置为false则其他连接也可以访问该队列。默认值为false。 autoDelete这是另一个布尔值参数表示队列是否自动删除。如果设置为true则当最后一个消费者取消订阅时队列将被删除如果设置为false则队列将一直存在直到手动删除或服务器重启。默认值为false。 arguments这是一个可选参数用于设置队列的其他属性比如消息的最大长度、最大优先级等。 channel.basicPublish 参数 exchange这是一个字符串参数代表交换机的名称。如果不需要使用特定的交换机可以传递一个空字符串。交换机是RabbitMQ中用于接收生产者发送的消息并根据绑定规则路由到队列的组件。 routingKey这也是一个字符串参数它指定了发布消息的队列。无论通道绑定到哪个队列最终发布的消息都会包含这个指定的路由键。路由键是用来确定消息应该发送到哪个队列的重要信息。 message这是要发布的消息本身通常是字节数组的形式。 properties这是一个可选参数用于设置消息的属性比如消息的优先级、过期时间等。 在使用channel.basicPublish时需要注意以下几点 exchange和routingKey不能为空在AMQImpl类中的实现要求这两个参数都不能为null否则会抛出异常。 交换机类型根据不同的需求可以选择不同类型的交换机如fanout、direct或topic。每种类型的交换机都有其特定的路由规则。 非命名队列在某些情况下比如日志系统可以使用非命名队列这样消费者可以接收到所有相关的日志消息而不是特定的部分。 2. 消费者代码 public class SingleConsumer {private final static String QUEUE_NAME hello;public static void main(String[] argv) throws Exception {ConnectionFactory factory new ConnectionFactory();factory.setHost(localhost);Connection connection factory.newConnection();Channel channel connection.createChannel();//声明队列同一个消息队列参数必须一致channel.queueDeclare(QUEUE_NAME, false, false, false, null);//定义了如何处理消息DeliverCallback deliverCallback (consumerTag, delivery) - {String message new String(delivery.getBody(), UTF-8);System.out.println( [x] Received message );};//接收、消费消息channel.basicConsume(QUEUE_NAME, true, deliverCallback, consumerTag - { });System.out.println( [] Waiting for messages. To exit press CTRLC);} } channel.basicConsume 参数 queue这是一个字符串参数代表要消费的队列的名称。如果队列不存在则会抛出异常。onMessage这是一个回调函数当有新的消息到达时会被调用。该函数需要接收两个参数一个表示消息内容的Delivery对象和一个表示通道的Channel对象。consumerTag这是一个可选参数用于标识消费者。如果没有指定则会自动生成一个唯一的标识符。autoAck这是一个布尔值参数表示是否自动确认消息。如果设置为true则在消息被处理后会自动发送确认信息如果设置为false则需要手动发送确认信息。默认值为false。arguments这是一个可选参数用于设置消费者的其他属性比如消息的最大长度、最大优先级等。 在使用channel.basicConsume时需要注意以下几点 队列名称队列名称应该是唯一的否则会抛出异常。消息处理在onMessage回调函数中需要对消息进行处理并根据需要发送确认信息。消费者标识符可以通过设置consumerTag来标识消费者以便在后续操作中进行识别和管理。消费者属性可以通过设置消费者的其他属性来控制消费者的行为比如设置消息的最大长度、最大优先级等。 多消费者 多个消费者比如一个工厂生产商品一个商店卖不完分给多个商店一起卖 生产者代码和上面一样 public class MultiProducer {private static final String TASK_QUEUE_NAME multi_queue;public static void main(String[] argv) throws Exception {//创建连接ConnectionFactory factory new ConnectionFactory();//设置本地连接factory.setHost(localhost);//创建队列创建频道类似客户端try (Connection connection factory.newConnection();Channel channel connection.createChannel()) {//队列持久化channel.queueDeclare(TASK_QUEUE_NAME, true, false, false, null);//设置消息Scanner scanner new Scanner(System.in);while(scanner.hasNext()){//输入消息String message scanner.nextLine();//发送消息channel.basicPublish(, TASK_QUEUE_NAME,MessageProperties.PERSISTENT_TEXT_PLAIN,message.getBytes(UTF-8));System.out.println( [x] Sent message );}}} } 控制处理任务的积压数最多同时处理任务数 channel.basicQos(1); //最多处理1个 消息确认机制 ack 确认、nack 消息失败、reject 拒绝 当消息拿走之后会有一个确认机制保证消息成功被消费。当消费者接收消息会给一个反馈确认消息的状态成功消息才会被移除。 支持配置 autoack 建议修改为 false根据实际情况手动确认。 //手动确认 channel.basicAck(delivery.getEnvelope().getDeliveryTag(),false); //手动拒绝 channel.basicNack(delivery.getEnvelope().getDeliveryTag(),false,false); 消费者代码 public class MultiConsumer {private static final String TASK_QUEUE_NAME multi_queue;public static void main(String[] argv) throws Exception {//创建连接ConnectionFactory factory new ConnectionFactory();factory.setHost(localhost);for (int i 0; i 2; i) {final Connection connection factory.newConnection();final Channel channel connection.createChannel();//队列持久化参数要一致channel.queueDeclare(TASK_QUEUE_NAME, true, false, false, null);System.out.println( [*] Waiting for messages. To exit press CTRLC);//控制处理任务的积压数最多同时处理任务数channel.basicQos(1);//定义了如何处理消息int finalI i;DeliverCallback deliverCallback (consumerTag, delivery) - {String message new String(delivery.getBody(), UTF-8);try {//处理工作的逻辑System.out.println( [x] Received 消费者 finalI 消息 message );//睡一定时间模拟机器处理能力有限Thread.sleep(20000);channel.basicAck(delivery.getEnvelope().getDeliveryTag(),false);} catch (InterruptedException e) {throw new RuntimeException(e);} finally {System.out.println( [x] Done);channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);}};//接收消息消费消息开启消息监听channel.basicConsume(TASK_QUEUE_NAME, false , deliverCallback, consumerTag - {});}} }