消息中间件二之AMQP实战
准备
环境
RabbitMQ server 3.7.16
RabbitMQ client 5.7.3
客户端代码使用的是RabbitMQ官网教程, 如下:
public static void main(String[] args) throws IOException, TimeoutException { ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
try (Connection connection = factory.newConnection();
Channel channel = connection.createChannel();) {
boolean durable = true;
channel.queueDeclare(QUEUE_NAME, durable, false, false, null);
String message = String.join(" ", "dMessage.......");
channel.exchangeDeclare("mind", "direct");
channel.basicPublish("mind", "", MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes());
System.out.println(" [x] Sent "" + message + """);
}
}
下面是用wireshark抓包结果
image
我们后面以编码No来具体分析
抓包分析
image
1-6是tcp创建连接的三次握手步骤, 这里不做过多分析
image
7-24是amqp创建连接的过程, 我们可以对照前面的博客中的说明文档来分析这里. 每次一端向另一端发送信息, 另一端在接收到后都会发送一个ack表示接收到了.
image
1 在tcp连接创建后, 客户端会向服务端发送协议版本信息, 这里是amqp的0.9.1版本, 服务端会校验版本是否接受, 如果不符合要求会返回错误信息, 这里只有正确信息, 后面我们自己实现客户端的时候可以实现一个错误例子.
image
2 服务端校验协议通过后, 会向客户端发送创建连接请求Connection.Start, 客户端在准备好后会返回一个Connection.Start-Ok. 接着服务端发送Connection.Tune与客户端进行参数调试, 调试的内容有Channel最大数量, Frame最大长度等. 客户端在调试后发送Connection.Tune-OK. 这个阶段就是对连接的参数调试.
image
3 参数调试之后, 客户端请求服务端打开连接Connection.Open, 服务端打开之后会返回Connection.Open-Ok. Connection打开成功后, 客户端请求打开通道Channel.Open, 服务端打开之后返回Channel.Open-Ok. 至此连接创建成功.
image
4 连接创建成功之后, 客户端进行队列和exchange的声明, Queue.Declare -> Queue.Declare-Ok, Exchange.Declare -> Exchange.Declare-Ok.
image
5 有了Exchange后, 客户端向Exchange发送信息, 我们可以看到发送的Exchange, 和发送的内容
image
image
6 发送内容结束后, 客户端关闭, 先关闭通道Channel, 然后关闭Connection.
image
7 最后是tcp关闭连接
代码分析
下面我们从代码层面分析这个过程, 下面是一个总体的时序图, 大家可以参考
image
下面我们还是按照抓包中看到的顺序来分析代码
创建tcp连接
代码很简单
ConnectionFactory factory = new ConnectionFactory();factory.setHost("localhost");
Connection connection = factory.newConnection();
我们重点看factory.newConnection(); 顺着方法我们很快找到了AutorecoveringConnection的init()方法
public void init() throws IOException, TimeoutException { this.delegate = this.cf.newConnection();
this.addAutomaticRecoveryListener(delegate);
}
在this.cf.newConnection()中重点看下
FrameHandler frameHandler = factory.create(addr, connectionName()); RecoveryAwareAMQConnection conn = createConnection(params, frameHandler, metricsCollector);
conn.start();
metricsCollector.newConnection(conn);
return conn;
大家debug代码可以看到factory是SocketFrameHandlerFactory的实例, 所以create中的代码如下:
public FrameHandler create(Address addr, String connectionName) throws IOException { String hostName = addr.getHost();
int portNumber = ConnectionFactory.portOrDefault(addr.getPort(), ssl);
Socket socket = null;
try {
socket = createSocket(connectionName);
configurator.configure(socket);
socket.connect(new InetSocketAddress(hostName, portNumber),
connectionTimeout);
return create(socket);
} catch (IOException ioe) {
quietTrySocketClose(socket);
throw ioe;
}
}
这里我们可以看到java网络的底层代码Socket,
socket.connect(new InetSocketAddress(hostName, portNumber), connectionTimeout);
这句代码完成了tcp的连接的创建工作.
(准备看这里源码的时候就想着肯定有个地方在做这步操作, 但就是找不到, 最后一点一点debug找到的...)
创建Connection
在上一步的最后, 把socket对象封装到一个FrameHandler实例中, 从这里我们可以猜想, 后面所有消息的通信都跟这个FrameHandler分不开.
我们继续看, 返回之后
FrameHandler frameHandler = factory.create(addr, connectionName()); RecoveryAwareAMQConnection conn = createConnection(params, frameHandler, metricsCollector);
conn.start();
使用FrameHandler实例构造了一个Connection对象, 然后调用了start()方法, 实际调用的是父类AMQConnection方法, 这个也是整个连接过程的重点
这里代码比较长, 我们选择一些重要的一点一点看
initializeConsumerWorkService(); // 初始化工作线程initializeHeartbeatSender(); // 初始化心跳线程
// Make sure that the first thing we do is to send the header,// which should cause any socket errors to show up for us, rather
// than risking them pop out in the MainLoop
// 确保我们在最开始发送的信息头在发生错误的时候不会出现在MainLoop中
// 这个实体就是为了接收在发送给服务端版本后接收服务端的Connection.Start方法的
AMQChannel.SimpleBlockingRpcContinuation connStartBlocker =
new AMQChannel.SimpleBlockingRpcContinuation();
// We enqueue an RPC continuation here without sending an RPC
// request, since the protocol specifies that after sending
// the version negotiation header, the client (connection
// initiator) is to wait for a connection.start method to
// arrive.
// 我们这里没有通过发送请求获取响应, 是因为服务端在接收到版本信息后会主动发送信息
_channel0.enqueueRpc(connStartBlocker);
enqueueRpc里面如下, 就是在循环中等待服务端信息接收成功的通知
private void doEnqueueRpc(Supplier<RpcWrapper> rpcWrapperSupplier) { synchronized (_channelMutex) {
boolean waitClearedInterruptStatus = false;
while (_activeRpc != null) {
try {
_channelMutex.wait(); // 后面在接收到Connection.Start方法后会通知
} catch (InterruptedException e) { //NOSONAR
waitClearedInterruptStatus = true;
// No Sonar: we re-interrupt the thread later
}
}
if (waitClearedInterruptStatus) {
Thread.currentThread().interrupt();
}
// 获取到通知后更新实体信息
_activeRpc = rpcWrapperSupplier.get();
}
}
_frameHandler.sendHeader(); // 发送版本信息, 对应抓包7
this._frameHandler.initialize(this); // 初始化, 主要是启动了一个MainLoop线程用于获取服务端信息
MainLoop线程的核心代码代码
Frame frame = _frameHandler.readFrame();readFrame(frame);
_frameHandler.readFrame() 内部代码如下, 这里大家可以查看译文中的2.3.5 Frame Details 帧的细节部分, 对照客户端是如何构造的, Frame结构如下
image
public static Frame readFrom(DataInputStream is) throws IOException { int type;
int channel;
try {
type = is.readUnsignedByte(); // 一个字节的类型信息
} catch (SocketTimeoutException ste) {
// System.err.println("Timed out waiting for a frame.");
return null; // failed
}
if (type == "A") { // 这里是处理, 如果服务端不支持客户端的版本, 会发送支持的版本信息, 开头是"A"
/*
* Probably an AMQP.... header indicating a version
* mismatch.
*/
/*
* Otherwise meaningless, so try to read the version,
* and throw an exception, whether we read the version
* okay or not.
*/
protocolVersionMismatch(is); // 这里面会抛出异常
}
channel = is.readUnsignedShort(); // 两个个字节的channel编号
int payloadSize = is.readInt(); // 4个字节的payload大小
byte[] payload = new byte[payloadSize];
is.readFully(payload); // 读取payloadSize大小的字节
int frameEndMarker = is.readUnsignedByte(); // 一个字节的尾部
if (frameEndMarker != AMQP.FRAME_END) {
throw new MalformedFrameException("Bad frame end marker: " + frameEndMarker);
}
// 构造对象并返回
return new Frame(type, channel, payload);
}
上一步主要是对信息的封装, 下面是客户端对封装对象的处理
private void readFrame(Frame frame) throws IOException { if (frame != null) {
_missedHeartbeats = 0;
if (frame.type == AMQP.FRAME_HEARTBEAT) {
// Ignore it: we"ve already just reset the heartbeat counter.
} else {
if (frame.channel == 0) { // the special channel 0通道是在创建连接过程中使用的
_channel0.handleFrame(frame); // 这一步就是将Connection.Start内容放到了channel提前设置的实体中
} else {
if (isOpen()) {
// If we"re still _running, but not isOpen(), then we
// must be quiescing, which means any inbound frames
// for non-zero channels (and any inbound commands on
// channel zero that aren"t Connection.CloseOk) must
// be discarded.
ChannelManager cm = _channelManager;
if (cm != null) {
ChannelN channel;
try {
channel = cm.getChannel(frame.channel);
} catch(UnknownChannelException e) {
// this can happen if channel has been closed,
// but there was e.g. an in-flight delivery.
// just ignoring the frame to avoid closing the whole connection
LOGGER.info("Received a frame on an unknown channel, ignoring it");
return;
}
channel.handleFrame(frame);
}
}
}
}
} else {
// Socket timeout waiting for a frame.
// Maybe missed heartbeat.
handleSocketTimeout();
}
}
我们回到start()方法中, 获取Connection.Start方法, 然后设置一些服务单发过来的参数
connStart = (AMQP.Connection.Start) connStartBlocker.getReply(handshakeTimeout/2).getMethod();
然后按照是响应Start.Ok, Tune方法, 对应抓包9-16
do { Method method = (challenge == null)
? new AMQP.Connection.StartOk.Builder()
.clientProperties(_clientProperties)
.mechanism(sm.getName())
.response(response)
.build()
: new AMQP.Connection.SecureOk.Builder().response(response).build();
try {
Method serverResponse = _channel0.rpc(method, handshakeTimeout/2).getMethod();
if (serverResponse instanceof AMQP.Connection.Tune) {
connTune = (AMQP.Connection.Tune) serverResponse;
} else {
challenge = ((AMQP.Connection.Secure) serverResponse).getChallenge();
response = sm.handleChallenge(challenge, username, password);
}
} catch (ShutdownSignalException e) {
Method shutdownMethod = e.getReason();
if (shutdownMethod instanceof AMQP.Connection.Close) {
AMQP.Connection.Close shutdownClose = (AMQP.Connection.Close) shutdownMethod;
if (shutdownClose.getReplyCode() == AMQP.ACCESS_REFUSED) {
throw new AuthenticationFailureException(shutdownClose.getReplyText());
}
}
throw new PossibleAuthenticationFailureException(e);
}
} while (connTune == null);
获取到调试信息, 设置本地参数
int channelMax = negotiateChannelMax(this.requestedChannelMax, connTune.getChannelMax());
_channelManager = instantiateChannelManager(channelMax, threadFactory);
int frameMax =
negotiatedMaxValue(this.requestedFrameMax,
connTune.getFrameMax());
this._frameMax = frameMax;
int heartbeat =
negotiatedMaxValue(this.requestedHeartbeat,
connTune.getHeartbeat());
setHeartbeat(heartbeat); 启动心跳线程
发送调整完毕方法TuneOk, 并请求打开连接Open
_channel0.transmit(new AMQP.Connection.TuneOk.Builder() .channelMax(channelMax)
.frameMax(frameMax)
.heartbeat(heartbeat)
.build());
_channel0.exnWrappingRpc(new AMQP.Connection.Open.Builder()
.virtualHost(_virtualHost)
.build());
至此Connection的连接已经创建并打开
创建Channel
接下来是Channel的创建, 我们前面代码中使用的Channel是特殊的, 专门用于创建Connection, 下面创建的是为了后面发送队列消息使用的Channel.
Channel channel = connection.createChannel() // 入口
根据AMQP文档, 创建Channel需要客户端发送Channel.Open方法然后接收服务端的Channel.OpenOk, 我们从抓包中也可以观察到. 我们一步一步跟踪代码, 代码层级比较深, 这里给出调用逻辑, 从下到上(对, 没错, 这就是创建Channel报错日志截取了部分)
com.rabbitmq.client.impl.AMQChannel.privateRpc(AMQChannel.java:295)com.rabbitmq.client.impl.AMQChannel.exnWrappingRpc(AMQChannel.java:141)
com.rabbitmq.client.impl.ChannelN.open(ChannelN.java:133)
com.rabbitmq.client.impl.ChannelManager.createChannel(ChannelManager.java:182)
com.rabbitmq.client.impl.AMQConnection.createChannel(AMQConnection.java:555)
com.rabbitmq.client.impl.recovery.AutorecoveringConnection.createChannel(AutorecoveringConnection.java:165)
privateRpc的代码我们看下
private AMQCommand privateRpc(Method m) throws IOException, ShutdownSignalException{ SimpleBlockingRpcContinuation k = new SimpleBlockingRpcContinuation(m);
rpc(m, k); // 发送Channel.Open 方法
// At this point, the request method has been sent, and we
// should wait for the reply to arrive.
// 这里我们已经发送了请求, 我们应该等待响应
// Calling getReply() on the continuation puts us to sleep
// until the connection"s reader-thread throws the reply over
// the fence or the RPC times out (if enabled)
// 调用getReply()方法会阻塞直到获取到结果或者超时
if(_rpcTimeout == NO_RPC_TIMEOUT) {
return k.getReply();
} else {
try {
return k.getReply(_rpcTimeout);
} catch (TimeoutException e) {
throw wrapTimeoutException(m, e);
}
}
}
接收Channel.OpenOk方法是由MainLoop线程完成的, 方式类似之前获取Connection.Start方法.
消息发送
至此AMQP连接算是完全创建完毕, 下面就是消息队列相关. 首先是队列和Exchange的声明, 这里队列的声明其实没有什么用, 代码这么写就是为了看下声明过程
channel.queueDeclare(QUEUE_NAME, durable, false, false, null);channel.exchangeDeclare("mind", "direct");
channel.basicPublish("mind", "", MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes());
这里声明的方式非常简单大家跟着代码很容易明白, queue和Exchange的声明过程基本一样, 不同的是queue在声明前会校验下队列的合法性(长度). 它们获取响应结果的方式和Channel.OpenOk的获取方式一毛一样.
消息的发送过程也是发送一个AMQPCommand, 但是细节很多, 准备在后面实现客户端的部分再详细看.
关闭连接
程序执行结束, 执行try-with-resources部分, 自动执行close()方法, 执行顺序从下到上, 也就是先执行Channel的close(), 然后Connection的close(); 从抓包中也可以看到先发送的Channel.close方法, 再发送Connection.close方法. 代码细节的部分这里就不展开了, 会放到后面代码实现上.
总结
整体过了一遍主要流程, 后面我们会自己实现一个简单客户端加深下理解; 这个过程中除了了解了客户端的操作流程外, 对java的部分知识也学习了一下
try-with-resources 在关闭时, 执行关闭的顺序和声明顺序相反;
try-with-resources 也可以有catch和finally块, 它们是在try-with-resources声明关闭之后执行的.
java线程状态流转
image
客户端实现(待完成~)
今天我们的目标是实现rabbitmq客户端, 并使用该客户端发送消息到指定Exchange中.
tcp连接创建
超级简单
socket = new Socket();socket.connect(new InetSocketAddress(host, port));
// 保存连接输入输出流
inputStream = new DataInputStream(new BufferedInputStream(socket.getInputStream()));
outputStream = new DataOutputStream(new BufferedOutputStream(socket.getOutputStream()));
抓包
image
发送信息头
我们通过抓包和源码知道, 发送头就是发送了"AMQP0091"
private int major = 0;private int minor = 9;
private int revision = 1;
outputStream.write("AMQP".getBytes());outputStream.write(0);
outputStream.write(major);
outputStream.write(minor);
outputStream.write(revision);
outputStream.flush();
抓包结果
image
可以看到服务端已经认可协议, 并发送了Connection.Start方法过来.
如果我们发送的协议服务端不认识会怎么样, 我们把major改为2试试
抓包结果
image
自己看下我们发的内容如下
image
我们是发送了0291, 抓包是支持AMQP协议的, 所以这里应该是不认识了, 所以显示为unknown version, 但是让我不理解的是服务端返回的结果也是unknown version, 根据AMQP文档中的说明, 服务端这时应该返回支持的协议, 我们点开看下
image
的确是0091正常的协议, 但是抓包软件没有显示出来, 很奇怪~
Connection.StartOk
以上是 消息中间件二之AMQP实战 的全部内容， 来源链接： utcz.com/z/513729.html
