rabbitmq（一）简单队列、轮询分发、公平分发

作者: guideEmotion | 来源:发表于2019-05-20 12:49 被阅读0次

rabbitmq（一）简单队列、轮询分发、公平分发
【rabbitMQ】rabbitMQ的队列、消息持久化
RabbitMQ轮询分发和公平分发
12.消费方：消息分发，消息顺序性
rabbit入门（三）工作队列轮询分发和公平分发
RabbitMQ-工作队列
springboot2.0集成rabbitmq
工作队列之轮询分发
消息队列Rabbitmq的交换器类型
Celery和RabbitMQ

一基本介绍

1.1 可以解决什么问题

异步处理
应用解耦
流量削峰
日志处理

1.2 安装

学习就装windows版本，方便测试

https://www.rabbitmq.com官网，下载windows版本。需要先安装一个erlang的环境。
先安装环境，后安装rabbitmq.一直next即可。
打开rabbitmq命令行，然后输命令

rabbitmq-plugins enable rabbitmq_management

浏览器打开 http://localhost:15672/ 账号/密码都是guest

1.3 用户管理

image.png

1.4 virtual hosts

virtual hosts 相当于数据库的db
1.先新增

点击

2.点击新增，进行配置。增加一个用户

image.png

二简单队列

模型

image.png
P：消息的生产者
红色的：消息队列
C：消费者

2.1 创建项目

1.用idea创建一个项目，添加依赖

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>com.zyc</groupId>
    <artifactId>myrabbit1</artifactId>
    <version>1.0-SNAPSHOT</version>

    <dependencies>
        <dependency>
            <groupId>com.rabbitmq</groupId>
            <artifactId>amqp-client</artifactId>
            <version>5.6.0</version>
        </dependency>

        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-api</artifactId>
            <version>1.7.10</version>
        </dependency>

        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-log4j12</artifactId>
            <version>1.7.5</version>
            <scope>test</scope>
        </dependency>

        <dependency>
            <groupId>log4j</groupId>
            <artifactId>log4j</artifactId>
            <version>1.2.17</version>
        </dependency>

        <dependency>
            <groupId>junit</groupId>
            <artifactId>junit</artifactId>
            <version>4.12</version>
            <scope>test</scope>
        </dependency>


    </dependencies>

</project>

2.2 获取连接

import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

/**
 * @author zhuyc
 * @create 2019-05-08 22:02
 */
public class ConnectiionUtil {

    public static Connection getConnection() throws IOException, TimeoutException {

        ConnectionFactory factory = new ConnectionFactory();

        factory.setHost("127.0.0.1");

        factory.setPort(5672);

        factory.setVirtualHost("/vhost_zhuyc");

        factory.setUsername("zhuyc");

        factory.setPassword("123456");
        return factory.newConnection();
    }

}

2.3 发送消息

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.zhuyc.util.ConnectiionUtil;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

/**
 * @author zhuyc
 * @create 2019-05-08 22:15
 */
public class Sender {

    private static  final String QUEUE_NAME = "test_simple_queue";
    
    public static void main(String[] args) throws IOException, TimeoutException {

        
        //获取连接
        Connection connection = ConnectiionUtil.getConnection();

        //从连接中获取一个通道
        Channel channel = connection.createChannel();
        //创建队列声明
        channel.queueDeclare(QUEUE_NAME,false,false,false,null);
        
        String msg = "hello world!simple queue";
        
        channel.basicPublish("",QUEUE_NAME,null,msg.getBytes());

        System.out.println("send end");
        channel.close();
        connection.close();
        
        
    }
    
}

2.4 消息接收

import com.rabbitmq.client.*;
import com.zhuyc.util.ConnectiionUtil;
import com.rabbitmq.client.AMQP.BasicProperties;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

/**
 * @author zhuyc
 * @create 2019-05-08 22:27
 */
public class Receiver
{

    private static  final String QUEUE_NAME = "test_simple_queue";

    public static void main(String[] args) throws IOException, TimeoutException {

        //获取连接
        Connection connection = ConnectiionUtil.getConnection();

        //从连接中获取一个通道
        Channel channel = connection.createChannel();

        DefaultConsumer consumer = new DefaultConsumer(channel) {


            public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body)
                    throws IOException {
                super.handleDelivery(consumerTag, envelope, properties, body);
                // 接收到的消息
                String message = new String(body);
                System.out.println("Receive 接收到的消息 " + message);
            }


        };
        channel.basicConsume(QUEUE_NAME,true,consumer);
    }

}

这里视频中的rabbitMQ client的版本较低，和新版的api不一样。需要看对应版本的官方文档
参考：https://blog.csdn.net/qq_37653556/article/details/82753890

2.5 简单队列的不足

耦合性高，生产者一一对应消费者（如果想要多个消费者消费队列中的消息，这时候就不行了）。
队列名变更。这时候得同时变更

三 Work queues 工作队列

模型

image.png

代码和上面是一样，只是多了一个消费者去监听队列

2.1 轮询分发

消费者1

package com.zhuyc.work;

import com.rabbitmq.client.*;
import com.zhuyc.util.ConnectiionUtil;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

/**
 * @author zhuyc
 * @create 2019-05-11 16:24
 */
public class Receiver1 {
    private static  final String QUEUE_NAME = "test_work_queue";

    public static void main(String[] args) throws IOException, TimeoutException {

        //获取连接
        Connection connection = ConnectiionUtil.getConnection();

        //从连接中获取一个通道
        Channel channel = connection.createChannel();

        channel.queueDeclare(QUEUE_NAME,false,false,false,null);
        DefaultConsumer consumer = new DefaultConsumer(channel) {

            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body)
                    throws IOException {
                super.handleDelivery(consumerTag, envelope, properties, body);
                // 接收到的消息
                String message = new String(body);
                System.out.println("Receive1 接收到的消息 " + message);

                try {
                    Thread.sleep(2000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }finally {
                    System.out.println("Receive1 done" );
                }
            }
        };
        channel.basicConsume(QUEUE_NAME,true,consumer);
    }
}

消费者2

public static void main(String[] args) throws IOException, TimeoutException {

        //获取连接
        Connection connection = ConnectiionUtil.getConnection();

        //从连接中获取一个通道
        Channel channel = connection.createChannel();

        channel.queueDeclare(QUEUE_NAME,false,false,false,null);

        DefaultConsumer consumer = new DefaultConsumer(channel) {


            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body)
                    throws IOException {
                super.handleDelivery(consumerTag, envelope, properties, body);
                // 接收到的消息
                String message = new String(body);
                System.out.println("Receive2 接收到的消息 " + message);
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }finally {
                    System.out.println("Receive2 done" );
                }
            }


        };

现象
消费者1和消费者处理得消息数量是一样的，你一个我一个
这种方式叫做轮询分发，就是不管哪个消费者效率较高，处理速度较快，平均分

2.2 公平分发

使用basicQos（prefetch=1）,即每次取完之后会对消息队列反馈。（注意：使用公平分发，必须关闭自动应答ack，改成手动）

生产者


    public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {

        Connection connection = ConnectiionUtil.getConnection();

        Channel channel = connection.createChannel();

        channel.queueDeclare(QUEUE_NAME,false,false,false,null);

        /**
         * 每个消费者发送确认消息之前，消息队列不发送下一条消息到消费者，一次只处理一个消息
         *
         * 限制发送给同一个消费者 不得超过一条消息
         */
        int prefetchCount = 1;
        channel.basicQos(prefetchCount);


        for (int i =0;i<50;i++){
            String msg = "hello"+i;
            channel.basicPublish("",QUEUE_NAME,null,msg.getBytes());
            Thread.sleep(30);
            System.out.println(i+" 发送完毕");


        }
        System.out.println("send end..");


    }

消费者

public static void main(String[] args) throws IOException, TimeoutException {

        //获取连接
        Connection connection = ConnectiionUtil.getConnection();

        //从连接中获取一个通道
        final Channel channel = connection.createChannel();

        channel.queueDeclare(QUEUE_NAME,false,false,false,null);

        channel.basicQos(1);//保证一次只分发一个

        DefaultConsumer consumer = new DefaultConsumer(channel) {


            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body)
                    throws IOException {
                super.handleDelivery(consumerTag, envelope, properties, body);
                // 接收到的消息
                String message = new String(body);
                System.out.println("Receive1 接收到的消息 " + message);

                try {
                    Thread.sleep(2000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }finally {
                    System.out.println("Receive1 done" );
                    //手动回执
                    channel.basicAck(envelope.getDeliveryTag(),false);
                }
            }


        };
        boolean autoAck = false;//关闭自动应答ack
        channel.basicConsume(QUEUE_NAME,autoAck ,consumer);

现象
这样消费快的消费者就会消费更多的消息

参考
1.视频讲解

rabbitmq（一）简单队列、轮询分发、公平分发
一基本介绍 1.1 可以解决什么问题异步处理应用解耦流量削峰日志处理 1.2 安装学习就装window...
【rabbitMQ】rabbitMQ的队列、消息持久化
0 轮询，公平分发 1 不公平分发
RabbitMQ轮询分发和公平分发
轮询分发先看代码，生产者生产十条消息。开启两个消费者，组成工作队列，消费者1消费完一条消息后将线程挂起1秒，消费...
12.消费方：消息分发，消息顺序性
1.消息分发当RabbitMQ队列拥有多个消费者时，队列收到的消息将以轮询的方式分发给消费者。每条消息只会发给订...
rabbit入门（三）工作队列轮询分发和公平分发
rabbit是一个消息代理和队列服务器。在实际开发当中，生产者发送消息是简单的，只需将消息放入消息中间件的时候就好...
RabbitMQ-工作队列
work queues 工作队列轮询分发为什么会出现工作队列simple队列是一一对应的，实际开发中，...
springboot2.0集成rabbitmq
安装rabbitmq 简介: rabbitmq即一个消息队列,主要用来实现应用程序的异步和解耦,消息缓冲,消息分发...
工作队列之轮询分发
直接上代码生产者1 消费者1 上述代码运行可以发现，即使消费者二处理的时间更快，但是消费者二并没有消费更多的消息，...
消息队列Rabbitmq的交换器类型
一、交换器类型在rabbitmq中，生产者的消息都是通过交换器来接收，然后再从交换器分发到不同的队列中去，在分发...
Celery和RabbitMQ
RabbitMQ是个消息队列的实现。将消息队列应用到何种场景，得看自己需求。你可以拿它自己实现任务分发（自己实现...

rabbitmq（一）简单队列、轮询分发、公平分发

一基本介绍

1.1 可以解决什么问题

1.2 安装

1.3 用户管理

1.4 virtual hosts

二简单队列

2.1 创建项目

2.2 获取连接

2.3 发送消息

2.4 消息接收

2.5 简单队列的不足

三 Work queues 工作队列

2.1 轮询分发

2.2 公平分发

相关文章

rabbitmq（一）简单队列、轮询分发、公平分发

【rabbitMQ】rabbitMQ的队列、消息持久化

RabbitMQ轮询分发和公平分发

12.消费方：消息分发，消息顺序性

rabbit入门（三）工作队列轮询分发和公平分发

RabbitMQ-工作队列

springboot2.0集成rabbitmq

工作队列之轮询分发

消息队列Rabbitmq的交换器类型

Celery和RabbitMQ

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读

rabbitmq（一）简单队列、轮询分发、公平分发

一 基本介绍

1.1 可以解决什么问题

1.2 安装

1.3 用户管理

1.4 virtual hosts

二 简单队列

2.1 创建项目

2.2 获取连接

2.3 发送消息

2.4 消息接收

2.5 简单队列的不足

三 Work queues 工作队列

2.1 轮询分发

2.2 公平分发

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读

一基本介绍

二简单队列