[一]RabbitMQ-客户端源码之ConnectionFactory

简介: 首先看一段amqp-client发送端的示例代码(展示出主要部分):ConnectionFactory factory = new ConnectionFactory();factory.

首先看一段amqp-client发送端的示例代码(展示出主要部分):

ConnectionFactory factory = new ConnectionFactory();
factory.setHost(ip);
factory.setPort(5672);
factory.setUsername("root");
factory.setPassword("root");
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();
String message = "RabbitMQ Demo Test:" + System.currentTimeMillis();
channel.basicPublish(EXCHANGE_NAME, routingKey, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes());
channel.close();
connection.close();

相信使用rabbitmq java客户端的同学来说,这段代码并不陌生,主要的作用是发送一条消息至broker然后关闭。通过wireshark抓包工具可以看到整个AMQP协议的流程,如下图:
(10.101.48.240是client的ip,10.198.197.73是broker的ip)

这里写图片描述

下面通过源码来分析下Connection有关的整个流程,对于上面AMQP流程中的Protocol-Header到Connection.Open-Ok的部分。

首先是ConnectionFactory类(文章开篇的demo中),这里主要包含一些与broker连接的配置参数等,比如:username, password, virtualHost, host,port, requestedChannelMax, requestedFrameMax, requestedHeartbeat, connectionTimeout, shutdownTimeout(只列出部分)。

这个类中其余都是些Getter和Setter方法,但是有个newConnection方法是关键,文中开篇的demo代码下面列出详细内容:

/**
 * Create a new broker connection, picking the first available address from
 * the list.
 *
 * If <a href="http://www.rabbitmq.com/api-guide.html#recovery">automatic connection recovery</a>
 * is enabled, the connection returned by this method will be {@link Recoverable}. Future
 * reconnection attempts will pick a random accessible address from the provided list.
 *
 * @param executor thread execution service for consumers on the connection
 * @param addrs an array of known broker addresses (hostname/port pairs) to try in order
 * @return an interface to the connection
 * @throws java.io.IOException if it encounters a problem
 * @see <a href="http://www.rabbitmq.com/api-guide.html#recovery">Automatic Recovery</a>
 */
public Connection newConnection(ExecutorService executor, Address[] addrs)
        throws IOException, TimeoutException {
    FrameHandlerFactory fhFactory = createFrameHandlerFactory();
    ConnectionParams params = params(executor);

    if (isAutomaticRecoveryEnabled()) {
        // see com.rabbitmq.client.impl.recovery.RecoveryAwareAMQConnectionFactory#newConnection
        AutorecoveringConnection conn = new AutorecoveringConnection(params, fhFactory, addrs);
        conn.init();
        return conn;
    } else {
        IOException lastException = null;
        for (Address addr : addrs) {
            try {
                FrameHandler handler = fhFactory.create(addr);
                AMQConnection conn = new AMQConnection(params, handler);
                conn.start();
                return conn;
            } catch (IOException e) {
                lastException = e;
            }
        }
        throw (lastException != null) ? lastException : new IOException("failed to connect");
    }
}

方法中首先是FrameHandlerFactory fhFactory = createFrameHandlerFactory();这个是用来处理client与broker之间的通信帧(Frame)的,包括建立通信链路(java的原生socket,注意这里没有NIO也没有netty)。

protected FrameHandlerFactory createFrameHandlerFactory() throws IOException {
    return new FrameHandlerFactory(connectionTimeout, factory, socketConf, isSSL());
}

调用createFrameHandlerFactory()方法得到FrameHandlerFactory对象之后再:“ FrameHandler handler = fhFactory.create(addr);”返回的是SocketFrameHandler对象,这个对象是对Socket的一个封装,完全可以看成是一个Socket对象。
注意这里的Socket的TCP_NODELAY参数默认设置为true,而不是默认的false。当然你也可以调用ConnectionFactory的setSocketConfigurator自行设置。

//这个方法是DefaultSocketConfigurator的唯一的方法
public void configure(Socket socket) throws IOException {
    // disable Nagle's algorithm, for more consistently low latency
    socket.setTcpNoDelay(true);
}

有关Socket的TCP_NODELAY参数:默认情况下发送数据是采用Negale算法。Negale算法是指发送方数据不会立刻发送出去,而是先放在缓冲区内,等待缓冲区满了,在发出去。Negale算法适用于需要发送大量数据的应用场景。这种算法减少传输的次数增加性能。但是如果对于需要即使响应的,小批量数据的应用场景,例如网络游戏就不能采用Negale算法了。默认是false,表示采用Negale算法。

ConnectionParams 主要用来配置与broker连接相关的参数,比如username,password,vhost等。这个与前面Socket的参数不同,需要注意区分。

之后if(isAutomaticRecoveryEnabled()){}之内的方法是建立可自动恢复连接的,这个可以忽略,直接看else里面的代码,因为if和else大体功能上一致,else里的更通用一些,也是默认的。上面提到 FrameHandler handler = fhFactory.create(addr);这段代码返回的是SocketFrameHandler对象,之后: AMQConnection conn = new AMQConnection(params, handler);这句通过参数和Socket与broker建立连接。之后初始化:conn.start();完成之后客户端就已经和broker建立了正常的连接了.

有关AMQConnection的详细内容将在下一篇文章[二]RabbitMQ-客户端源码之AMQConnection中讲述。


附:本系列全集

  1. [Conclusion]RabbitMQ-客户端源码之总结
  2. [一]RabbitMQ-客户端源码之ConnectionFactory
  3. [二]RabbitMQ-客户端源码之AMQConnection
  4. [三]RabbitMQ-客户端源码之ChannelManager
  5. [四]RabbitMQ-客户端源码之Frame
  6. [五]RabbitMQ-客户端源码之AMQChannel
  7. [六]RabbitMQ-客户端源码之AMQCommand
  8. [七]RabbitMQ-客户端源码之AMQPImpl+Method
  9. [八]RabbitMQ-客户端源码之ChannelN
  10. [九]RabbitMQ-客户端源码之Consumer
相关实践学习
消息队列RocketMQ版:基础消息收发功能体验
本实验场景介绍消息队列RocketMQ版的基础消息收发功能,涵盖实例创建、Topic、Group资源创建以及消息收发体验等基础功能模块。
消息队列 MNS 入门课程
1、消息队列MNS简介 本节课介绍消息队列的MNS的基础概念 2、消息队列MNS特性 本节课介绍消息队列的MNS的主要特性 3、MNS的最佳实践及场景应用 本节课介绍消息队列的MNS的最佳实践及场景应用案例 4、手把手系列:消息队列MNS实操讲 本节课介绍消息队列的MNS的实际操作演示 5、动手实验:基于MNS,0基础轻松构建 Web Client 本节课带您一起基于MNS,0基础轻松构建 Web Client
目录
相关文章
|
2月前
|
消息中间件 存储 监控
深度写作:深入源码理解MQ长轮询优化机制
【11月更文挑战第22天】在分布式系统中,消息队列(Message Queue, MQ)扮演着至关重要的角色。MQ不仅实现了应用间的解耦,还提供了异步消息处理、流量削峰等功能。而在MQ的众多特性中,长轮询(Long Polling)机制因其能有效提升消息处理的实时性和效率,备受关注。
89 12
|
2月前
|
消息中间件 存储 Java
深入源码理解MQ长轮询优化机制
【11月更文挑战第22天】在分布式系统中,消息队列(MQ)作为一种重要的中间件,广泛应用于解耦、异步处理、流量削峰等场景。其中,延时消息和定时消息作为MQ的高级功能,能够进一步满足复杂的业务需求。为了实现这些功能,MQ系统需要进行一系列优化,长轮询机制便是其中的关键一环。本文将深入探讨MQ如何设计延时消息和定时消息的优化机制,特别是长轮询机制的实现原理及其在Java中的模拟实现。
49 2
|
7月前
|
消息中间件 开发工具 RocketMQ
消息队列 MQ产品使用合集之如何关闭客户端的日志记录
消息队列(MQ)是一种用于异步通信和解耦的应用程序间消息传递的服务,广泛应用于分布式系统中。针对不同的MQ产品,如阿里云的RocketMQ、RabbitMQ等,它们在实现上述场景时可能会有不同的特性和优势,比如RocketMQ强调高吞吐量、低延迟和高可用性,适合大规模分布式系统;而RabbitMQ则以其灵活的路由规则和丰富的协议支持受到青睐。下面是一些常见的消息队列MQ产品的使用场景合集,这些场景涵盖了多种行业和业务需求。
|
6月前
|
消息中间件 Java C语言
消息队列 MQ使用问题之在使用C++客户端和GBase的ESQL进行编译时出现core dump,该怎么办
消息队列(MQ)是一种用于异步通信和解耦的应用程序间消息传递的服务,广泛应用于分布式系统中。针对不同的MQ产品,如阿里云的RocketMQ、RabbitMQ等,它们在实现上述场景时可能会有不同的特性和优势,比如RocketMQ强调高吞吐量、低延迟和高可用性,适合大规模分布式系统;而RabbitMQ则以其灵活的路由规则和丰富的协议支持受到青睐。下面是一些常见的消息队列MQ产品的使用场景合集,这些场景涵盖了多种行业和业务需求。
|
6月前
|
消息中间件 存储 监控
消息队列 MQ使用问题之客户端重启后仍然出现broker接收消息不均匀,该怎么办
消息队列(MQ)是一种用于异步通信和解耦的应用程序间消息传递的服务,广泛应用于分布式系统中。针对不同的MQ产品,如阿里云的RocketMQ、RabbitMQ等,它们在实现上述场景时可能会有不同的特性和优势,比如RocketMQ强调高吞吐量、低延迟和高可用性,适合大规模分布式系统;而RabbitMQ则以其灵活的路由规则和丰富的协议支持受到青睐。下面是一些常见的消息队列MQ产品的使用场景合集,这些场景涵盖了多种行业和业务需求。
|
7月前
|
消息中间件 数据可视化 Go
Rabbitmq 搭建使用案例 [附源码]
Rabbitmq 搭建使用案例 [附源码]
58 0
|
3月前
|
传感器 数据可视化 网络协议
DIY可视化整合MQTT生成UniApp源码
DIY可视化整合MQTT生成UniApp源码
65 0
|
6月前
|
消息中间件 安全 PHP
消息队列 MQ使用问题之如何获取PHP客户端代码
消息队列(MQ)是一种用于异步通信和解耦的应用程序间消息传递的服务,广泛应用于分布式系统中。针对不同的MQ产品,如阿里云的RocketMQ、RabbitMQ等,它们在实现上述场景时可能会有不同的特性和优势,比如RocketMQ强调高吞吐量、低延迟和高可用性,适合大规模分布式系统;而RabbitMQ则以其灵活的路由规则和丰富的协议支持受到青睐。下面是一些常见的消息队列MQ产品的使用场景合集,这些场景涵盖了多种行业和业务需求。
|
5月前
|
安全 网络性能优化
MQTT 客户端 MQTT.fx 使用说明
MQTT 客户端 MQTT.fx 使用说明
520 0
|
6月前
|
消息中间件 JavaScript Linux
消息队列 MQ操作报错合集之客户端在启动时遇到了连接错误,是什么原因
消息队列(MQ)是一种用于异步通信和解耦的应用程序间消息传递的服务,广泛应用于分布式系统中。针对不同的MQ产品,如阿里云的RocketMQ、RabbitMQ等,它们在实现上述场景时可能会有不同的特性和优势,比如RocketMQ强调高吞吐量、低延迟和高可用性,适合大规模分布式系统;而RabbitMQ则以其灵活的路由规则和丰富的协议支持受到青睐。下面是一些常见的消息队列MQ产品的使用场景合集,这些场景涵盖了多种行业和业务需求。