rabbitmq创建缓存连接工厂

转自：

【RabbitMQ-3】连接池的配置_小胖学编程的博客-CSDN博客文章目录1. rabbitmq的connection连接池1.1 问题提出1.1.1 Connection对象管理以及性能1.1.2 Channel对象管理以及性能1.2 Spring AMQP线程池配置1.2.1 ConnectionFactory连接工厂1.2.2 消费发送和接收使用不同的Connectionjava NIO是IO的多路复用，Channel连接是TCP的多路复用。那么他们有什么关系呢？NIO是服务器开启一个线程，在内核中使用select()进行轮询管理一些socket，当sockehttps://blog.csdn.net/qq_29595463/article/details/107858293

文章目录

1. rabbitmq的connection连接池
        1.1 问题提出
        1.1.1 Connection对象管理以及性能
            1.1.2 Channel对象管理以及性能
        1.2 Spring AMQP线程池配置
            1.2.1 ConnectionFactory连接工厂
            1.2.2 消费发送和接收使用不同的Connection

java NIO是IO的多路复用，Channel连接是TCP的多路复用。那么他们有什么关系呢？
NIO是服务器开启一个线程，在内核中使用select()进行轮询管理一些socket，当socket数据准备好时，会通知应用程序进行读写请求。系统之间那点事-NIO（内附IO模型）-IO/NIO/AIO到底是什么。服务器看起来就好像是一个socket在通信，实现了多路复用。
channel复用TCP连接，是为了避免TCP连接创建和销毁的性能损耗，而多个channel使用一个tcp连接。

1. rabbitmq的connection连接池

RabbitMQ.png

我们看到一个Connection里面可以包含多个channel。那么我们在连接broker时，connection和channel的关系是什么？
1.1 问题提出
1.1.1 Connection对象管理以及性能

Connection连接本质上就是TCP连接，系统之间那点事-问题驱动-TCP的连接和关闭是比较耗费时间的。我们可以使用一个单例的Connection对象创建多个Channel来实现数据传输，但是在channel信息比较大的情况下，Connection带宽会限制消息的传输。那么需要设计Connection池，将流量分摊到不同的connection上。
1.1.2 Channel对象管理以及性能

Channel对象的创建和销毁也是非常耗时的，推荐共享使用Channel，而不是每次都创建和销毁Channel。那如何设计一个channel线程池呢？

官网对于Connection的解读：

AMQP 0-9-1 connections are typically long-lived. AMQP 0-9-1 is an application level protocol that uses TCP for reliable delivery. Connections use authentication and can be protected using TLS. When an application no longer needs to be connected to the server, it should gracefully close its AMQP 0-9-1 connection instead of abruptly closing the underlying TCP connection.

大概意思就是：AMQP 0-9-1一般是一个TCP的长链接，当应用程序不再需要连接到服务器时，应该正常关闭AMQP 0-9-1连接而不是关闭TCP连接。

官网对于Channel的解读：

Some applications need multiple connections to the broker. However, it is undesirable to keep many TCP connections open at the same time because doing so consumes system resources and makes it more difficult to configure firewalls. AMQP 0-9-1 connections are multiplexed withchannels that can be thought of as “lightweight connections that share a single TCP connection”.
    Every protocol operation performed by a client happens on a channel. Communication on a particular channel is completely separate from communication on another channel, therefore every protocol method also carries a channel ID (a.k.a. channel number), an integer that both the broker and clients use to figure out which channel the method is for.
    A channel only exists in the context of a connection and never on its own. When a connection is closed, so are all channels on it.
    For applications that use multiple threads/processes for processing, it is very common to open a new channel per thread/process and not share channels between them.

一些应用需要同时创建多个连接到broker也就是RabbitMQ服务器上。然而因为防火墙的存在，很难同时创建多个连接。 AMQP 0-9-1连接使用多个channel连接实现对单一Connection的复用。
客户端的每一个协议操作都发送在channel上。每个协议方法携带者channel ID。broker和client使用channel ID来确定方法对应的channel。因此实现channel之间的数据隔离。
channel不能单独存在，仅存在connection上下文中。当connection关闭时，channel也会关闭。
多线程/进程之间打开一个channel但不共享channels是很普遍的。

通道和并发注意事项（线程安全）

As a rule of thumb, sharing Channel instances between threads is something to be avoided. Applications should prefer using a Channel per thread instead of sharing the same Channel across multiple threads.

线程之间共享channel是无法避免的，应用程序跟喜欢每个线程使用一个channel而不是跨线程共享相同的channel。

A classic anti-pattern to be avoided is opening a channel for each published message. Channels are supposed to be reasonably long-lived and opening a new one is a network round-trip which makes this pattern extremely inefficient.

要避免一个反例，为每一个发布的消息分配一个channel，开辟一个新的channel需要一个网络的往返，这种模式是很低效的。channel保持合理的存活时间。

It is possible to use channel pooling to avoid concurrent publishing on a shared channel: once a thread is done working with a channel, it returns it to the pool, making the channel available for another thread. Channel pooling can be thought of as a specific synchronization solution. It is recommended that an existing pooling library is used instead of a homegrown solution. For example, Spring AMQP which comes with a ready-to-use channel pooling feature.

（敲黑板，划重点）可以使用channel pool来避免共享channel上并发发布：一旦一个线程使用完了channel，那么它将返回到pool中。其他线程便可使用这个Channel。线程池是一个解决方案，可以使用 Spring AMQP线程池而不是自己开发。

__总结：__频繁建立TCP连接和channel连接是消耗性能的，于是我们希望可以共享connection或者channel。达到连接的复用。
1.2 Spring AMQP线程池配置

版本spring-rabbit:2.0.2.RELEASE
1.2.1 ConnectionFactory连接工厂

这个ConnectionFactory是Spring AMQP定义的连接工厂，负责创建连接。而CacheConnectionFactory实现支持对这些通道的缓存。

private static ConnectionFactory newRabbitConnectionFactory() {
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setAutomaticRecoveryEnabled(false);
        return connectionFactory;
    }

1
    2
    3
    4
    5

参数分析：

1. 开启confirm机制。
connectionFactory.setPublisherConfirms(true); connectionFactory.setPublisherReturns(true);
为了消息的不丢失，生产者可以设置事务或者confirm异步通知。但是事务性能并不是很好，所以一般使用confirm模式。
区别：（confirm保证达到交换机，return保证交换机到达队列）
如果消息没有到exchange,则confirm回调,ack=false；
如果消息到达exchange,则confirm回调,ack=true；
exchange到queue成功,则不回调return
exchange到queue失败,则回调return(需设置mandatory=true,否则不回回调,消息就丢了)

*注意：设置PublisherReturns状态为true，那么需要设置 rabbitTemplate.setMandatory(true);
具体如何保证消息不丢失，请参考RabbitMQ的消息不丢失机制

2. 配置模式
缓存模式一般两种：
connectionFactory.setCacheMode(CachingConnectionFactory.CacheMode.CONNECTION);

public static enum CacheMode {
CHANNEL,
CONNECTION;

private CacheMode() {
}
}

1
    2
    3
    4
    5
    6
    7

2.1 CHANNEL模式

程序运行期间ConnectionFactory只维护着一个connection，但是可以含有多个channel，操作rabbitmq之前必须先获取一个channel，否则将会阻塞。

相关参数配置：

connectionFactory.setConnectionCacheSize(3);
仅在CONNECTION模式下使用，指定connection缓存数量。

connectionFactory.setConnectionLimit(10);
仅在CONNECTION模式下使用，指定connection数量上限。

官网对于是否关闭channel解答：

Channels used within the framework (e.g. RabbitTemplate) will be reliably returned to the cache. If you create channels outside of the framework, (e.g. by accessing the connection(s) directly and invoking createChannel()), you must return them (by closing) reliably, perhaps in a finally block, to avoid running out of channels.

注意：若使用RabbitTemplate创建channel，那么无需关闭，但是自己新建connection创建channel，则需要手动关闭！避免channel溢出。

ConnectionFactory 代码：

@Bean
    public ConnectionFactory connectionFactory() {
        CachingConnectionFactory connectionFactory = new CachingConnectionFactory(host, port);
        connectionFactory.setUsername(username);
        connectionFactory.setPassword(password);
        //设置virtualHost。
        connectionFactory.setVirtualHost("/");
        //消息的确认机制（confirm）；
        connectionFactory.setPublisherConfirms(true);
        connectionFactory.setPublisherReturns(true);
        //setCacheMode：设置缓存模式，共有两种，CHANNEL和CONNECTION模式。
        //1、CONNECTION模式，这个模式下允许创建多个Connection，会缓存一定数量的Connection，每个Connection中同样会缓存一些Channel，
        // 除了可以有多个Connection，其它都跟CHANNEL模式一样。
        //2、CHANNEL模式，程序运行期间ConnectionFactory会维护着一个Connection，
        // 所有的操作都会使用这个Connection，但一个Connection中可以有多个Channel，
        // 操作rabbitmq之前都必须先获取到一个Channel，
        // 否则就会阻塞（可以通过setChannelCheckoutTimeout()设置等待时间），
        // 这些Channel会被缓存（缓存的数量可以通过setChannelCacheSize()设置）；
        connectionFactory.setCacheMode(CachingConnectionFactory.CacheMode.CONNECTION);   //设置CONNECTION模式，可创建多个Connection连接
        //设置每个Connection中缓存Channel的数量，不是最大的。操作rabbitmq之前（send/receive message等）
        // 要先获取到一个Channel.获取Channel时会先从缓存中找闲置的Channel，如果没有则创建新的Channel，
        // 当Channel数量大于缓存数量时，多出来没法放进缓存的会被关闭。
        connectionFactory.setChannelCacheSize(10);
        //单位：毫秒；配合channelCacheSize不仅是缓存数量，而且是最大的数量。
        // 从缓存获取不到可用的Channel时，不会创建新的Channel，会等待这个值设置的毫秒数
        //同时，在CONNECTION模式，这个值也会影响获取Connection的等待时间，
        // 超时获取不到Connection也会抛出AmqpTimeoutException异常。
        connectionFactory.setChannelCheckoutTimeout(600);

//仅在CONNECTION模式使用，设置Connection的缓存数量。
        connectionFactory.setConnectionCacheSize(3);
        //setConnectionLimit：仅在CONNECTION模式使用，设置Connection的数量上限。
        connectionFactory.setConnectionLimit(10);
        return connectionFactory;
    }

1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    29
    30
    31
    32
    33
    34
    35

RabbitTemplate 代码：

@Autowired
    @Bean
    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
        //客户端开启confirm模式
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        rabbitTemplate.setMandatory(true);
        rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
            @Override
            public void confirm(CorrelationData correlationData, boolean ack, String cause) {
                log.info("消息发送成功:correlationData({}),ack({}),cause({})", correlationData, ack, cause);
            }
        });
        rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {
            @Override
            public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
                log.info("消息丢失:exchange({}),route({}),replyCode({}),replyText({}),message:{}", exchange, routingKey, replyCode, replyText, message);
            }
        });
        return rabbitTemplate;
    }

1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20

1.2.2 消费发送和接收使用不同的Connection

当一个服务同时作为消息的发送端和接收端时，建议使用不同的Connection避免一方出现故障或者阻塞影响另一方。

只需要在RabbitTemplate中加入下面的配置，那么RabbitTemplate在创建Connection时，会根据这个boolean的值，选择使用ConnectionFactory本身或者ConnectionFactory中的publisherConnectionFactory（也即是一个ConnectionFactory）来创建。

rabbitTemplate.setUsePublisherConnection(true);

rabbitmq创建缓存连接工厂相关推荐

RabbitMQ创建远程连接用户
RabbitMQ不允许默认用户guest远程访问,所以需要创建一个新的用户,以下命令序列用于创建新用户admin(密码admin) D:\Program\RabbitMQ Server\rabbitm ...
[RabbitMQ]创建Java开发环境_消费者_生产者
我们将用 Java 编写两个程序.发送单个消息的生产者和接收消息并打印出来的消费者.我们将介绍 Java API 中的一些细节. 在下图中," P"是我们的生产者," C ...
自定义创建rabbitMQ的channel连接池
参考地址:https://blog.csdn.net/qq447995687/article/details/80233621 利用commons-pool2自定义对象池 commons-pool2是 ...
Socket 创建与连接
1. 定义 Socket 状态,SocketStatus.java /*** @Description: Socket 连接的状态*/public interface SocketStatus {/* ...
oracle创建DBLink连接
1.创建dblink的第一种方式,是在本地数据库tnsnames.ora文件中配置了要远程访问的数据库.tnsnames.ora文件在你安装oracle客户端安装文件里如:(E:\oracle\pr ...
mysql内连接和外连接的区别_Swoole4创建Mysql连接池
一 .什么是mysql连接池场景:每秒同时有1000个并发,但是这个mysql同时只能处理400个连接,mysql会宕机. 解决方案:连接池,这个连接池建立了200个和mysql的连接,这1000个 ...
SQLAlchemy的使用---外键ForeignKey数据库创建与连接
SQLAlchemy的使用---外键ForeignKey数据库创建与连接 # 一对多建表操作 from sqlalchemy.ext.declarative import declarative_ba ...
在Windows下创建硬连接和文件夹连接点的工具
之前提到过在Windows[NTFS]下面创建硬连接的命令行工具(请看<简化创建硬连接命令>),有没有更方便的方法呢?请看这次推荐的几个工具:(这三个工具都可以到http://schina ...
mysql 8.0创建远程连接用户
在虚拟机中,连接宿主机mysql数据库时,是无法直接连接,会出现如下错误. OperationalError: (1045, "Access denied for user 'root'@' ...
(0016)iOS 开发之Mac上Navicat Premium 创建远程连接和本地连接
1.下载安装 (百度云盘里面有安装文件和注册机) 链接: https://pan.baidu.com/s/1kVG1k71 密码: mr5g 破解教程看这篇博客:http://blog.csdn.ne ...

rabbitmq创建缓存连接工厂

rabbitmq创建缓存连接工厂相关推荐

最新文章

热门文章