Cobar启动完成，监听特定端口。整个认证的流程图： NIOAcceptor类继承自Thread类，该类的对象会以线程的方式运行，进行连接的监听。NIOAcceptor启动的初始化过程如下：1 、打开一个selector，获取一个ServerSocketChannel对象，对该对象的socket绑定特定的监
Cobar启动完成，监听特定端口。整个认证的流程图：
NIOAcceptor类继承自Thread类，该类的对象会以线程的方式运行，进行连接的监听。
NIOAcceptor启动的初始化过程如下：
1 、打开一个selector，获取一个ServerSocketChannel对象，对该对象的socket绑定特定的监听端口，并设置该channel为非阻塞模式，然后想selector注册该channel，绑定感兴趣的事件位OP_ACCEPT。
public NIOAcceptor(String name, int port, FrontendConnectionFactory factory) throws IOException {
super.setName(name);
this.port = port;
this.selector = Selector.open();
this.serverChannel = ServerSocketChannel.open();
//ServerSocket使用TCP
this.serverChannel.socket().bind(new InetSocketAddress(port));
this.serverChannel.configureBlocking(false);
this.serverChannel.register(selector, SelectionKey.OP_ACCEPT);
this.factory = factory;
}
2、 然后会启动该线程，线程的run函数如下：
public void run() {
final Selector selector = this.selector;
//线程一直循环
for (;;) {
++acceptCount;
try {
selector.select(1000L);
Set<SelectionKey> keys = selector.selectedKeys();
try {
for (SelectionKey key : keys) {
if (key.isValid() && key.isAcceptable()) {
//接受来自客户端的连接
accept();
} else {
key.cancel();
}
}
} finally {
keys.clear();
}
} catch (Throwable e) {
LOGGER.warn(getName(), e);
}
}
}
3 、 该线程会一直循环监听想该selector注册过的server channel所感兴趣的事件（OP_ACCEPT），当有新的连接请求时，selector就会返回，keys就是请求连接的所有的包含channel的key集合。
SelectionKey有如下属性：
- interest集合(使用&操作SelectionKey.OP_ACCEPT和key.interestOps())
- ready集合(key.readyOps()，可以使用&操作检测该集合，也可以使用is方法)
- Channel(key.channel())
- Selector(key.selector())
- 附加对象(key.attach(obj) ? Object obj = key.attachment())
private void accept() {
SocketChannel channel = null;
try {
//从服务器端获取管道,为一个新的连接返回channel
channel = serverChannel.accept();
//配置管道为非阻塞
channel.configureBlocking(false);
//前端连接工厂对管道进行配置，设置socket的收发缓冲区大小，TCP延迟等
//然后由成员变量factory的类型生产对于的类型的连接
//比如ServerConnectionFactory会返回ServerConnection实例，并对其属性进行设置
FrontendConnection c = factory.make(channel);
//设置连接属性
c.setAccepted(true);
c.setId(ID_GENERATOR.getId());
//从processors中选择一个NIOProcessor，将其和该连接绑定
NIOProcessor processor = nextProcessor();
c.setProcessor(processor);
//向读反应堆注册该连接，加入待处理队列
//select选择到感兴趣的事件后，会进行调用connection的read函数
processor.postRegister(c);
} catch (Throwable e) {
closeChannel(channel);
LOGGER.warn(getName(), e);
}
}
首先从serverchannel中accept后会返回一个socketchannel对象，然后设置该socket channel属性位非阻塞模式，然后将channel交给ServerConnectionFactory工厂，会产生一个ServerConnection对象。
FrontendConnectionFactory是一个抽象类，其中的getConnection方法是抽象方法，有具体子类连接工厂来实现。FrontendConnectionFactory的make方法对channel中的socket进行属性设置（接收和发送的缓冲区大小、延时、KeepAlive等），然后调用具体调用具体子类（ServerConnectionFactory）的getConnection来返回一个ServerConnection，返回后会在进行设置一下该ServerConnection的包头大小、最大包大小、设置连接的发送缓冲区队列、超时时间、字符编码，到此，工厂完成了新建连接的工作，返回一个连接的对象。返回后将该连接分配给一个processor，该processor会将该连接保存，processor也会对连接进行定期检查。
5、 processor还会向自己的reactorR进行注册该连接，加入reactorR的处理队列，并唤醒阻塞的select()方法。
反应堆中Reactor的R线程运行代码：
public void run() {
final Selector selector = this.selector;
for (;;) {
++reactCount;
try {
int res = selector.select();
LOGGER.debug(reactCount + ">>NIOReactor接受连接数:" + res);
register(selector);
Set<SelectionKey> keys = selector.selectedKeys();
try {
for (SelectionKey key : keys) {
Object att = key.attachment();
if (att != null && key.isValid()) {
int readyOps = key.readyOps();
if ((readyOps & SelectionKey.OP_READ) != 0) {
LOGGER.debug("select读事件");
read((NIOConnection) att);
} else if ((readyOps & SelectionKey.OP_WRITE) != 0) {
LOGGER.debug("select写事件");
write((NIOConnection) att);
} else {
key.cancel();
}
} else {
key.cancel();
}
}
} finally {
keys.clear();
}
} catch (Throwable e) {
LOGGER.warn(name, e);
}
}
}
该R线程也会一直循环运行，如果向该selector注册过的channel没有对应的感兴趣的事件发生，就会阻塞，直到有感兴趣的事件发生或被wakeup。返回后会运行register函数，将之前加入该reactor连接队列中的所有连接向该selector注册OP_READ事件。该注册的动作会调用Connection对象中的register方法进行注册，并像客户端主动发送握手数据包（HandshakePacket）。
?channel.register(selector, SelectionKey.OP_READ, this);注意最后一个this指针参数，表示将该连接作为附件，注册到selector，当有感兴趣的时间发生时，函数selector.selectedKeys()返回的SelectionKey集合中的对象中使用key.attachment()即可获取到上面注册时绑定的connection对象指针附件。目的就是为了通过该附件对象调用该连接类中定义的read函数来完成功能。如下所示：
private void read(NIOConnection c) {
try {
c.read();
} catch (Throwable e) {
c.error(ErrorCode.ERR_READ, e);
}
}
6、 连接类中定义的read函数定义在AbstractConnection类中。在该read函数 public void handle(final byte[] data) {
// 从线程池获取一个线程,异步处理前端数据
// 从processor中的线程池中获取一个可以执行的线程,执行Runnable任务
processor.getHandler().execute(new Runnable() {
@Override
public void run() {
try {
//调用具体NIOHandler子类的handle函数
handler.handle(data);
} catch (Throwable t) {
error(ErrorCode.ERR_HANDLE_DATA, t);
}
}
});
}
8、 handler在构造函数中初始化成前端认证处理器，用于处理前端权限认证。
public FrontendConnection(SocketChannel channel) {
super(channel);
.....................
//前端认证处理器
this.handler = new FrontendAuthenticator(this);
}
9、 由于Cobar是基于MySQL协议的，所以需要分析一下MySQL协议的具体格式。下面就先分析一下MySQL认证数据包的格式：
每个报文都分为消息头和消息体两部分，其中消息头是固定的四个字节，报文结构如下：
登录认证报文的报文数据部分格式如下：
10、 FrontendAuthenticator类对上面的数据包的具体处理如下：
- 读取信息到认证包对象
- 核对用户
- 核对密码
- 检查schema
public void handle(byte[] data) {
// check quit packet
if (data.length == QuitPacket.QUIT.length && data[4] == MySQLPacket.COM_QUIT) {
source.close();
return;
}
//新建认证包对象
AuthPacket auth = new AuthPacket();
//读取认证包到对象
auth.read(data);
// check user
if (!checkUser(auth.user, source.getHost())) {
failure(ErrorCode.ER_ACCESS_DENIED_ERROR, "Access denied for user '" + auth.user + "'");
return;
}
// check password
if (!checkPassword(auth.password, auth.user)) {
failure(ErrorCode.ER_ACCESS_DENIED_ERROR, "Access denied for user '" + auth.user + "'");
return;
}
// check schema
switch (checkSchema(auth.database, auth.user)) {
case ErrorCode.ER_BAD_DB_ERROR:
failure(ErrorCode.ER_BAD_DB_ERROR, "Unknown database '" + auth.database + "'");
break;
case ErrorCode.ER_DBACCESS_DENIED_ERROR:
String s = "Access denied for user '" + auth.user + "' to database '" + auth.database + "'";
failure(ErrorCode.ER_DBACCESS_DENIED_ERROR, s);
break;
default:
//认证成功,向客户端发送认证结果消息
success(auth);
}
}
在上面的auth.read函数中会按9中的协议格式进行读取数据到auth对象。认证成功后会执行：
protected void success(AuthPacket auth) {
//认证通过,设置连接属性:已认证\用户\数据库\处理器
source.setAuthenticated(true);
source.setUser(auth.user);
source.setSchema(auth.database);
source.setCharsetIndex(auth.charsetIndex);
//设置该连接的连接处理器为前端命令处理器
source.setHandler(new FrontendCommandHandler(source));
.......
ByteBuffer buffer = source.allocate();
source.write(source.writeToBuffer(AUTH_OK, buffer));
}
可以看到，在上面的函数中，设置连接对象source中的成员（是否认证、用户、数据库、编码、处理该连接后续数据包的处理器【handle方法】）
然后回复认证成功的消息。后面客户端再发送消息，会交给前端命令处理器进行处理。
客户端进行链接的时候Cobar服务器的输出：
16:59:19,388 INFO =============================================== 16:59:19,389 INFO Cobar is ready to startup ... 16:59:19,389 INFO Startup processors ... 16:59:19,455 INFO Startup connector ... 16:59:19,460 INFO Initialize dataNodes ... 16:59:19,506 INFO dnTest1:0 init success 16:59:19,514 INFO dnTest3:0 init success 16:59:19,517 INFO dnTest2:0 init success 16:59:19,527 INFO CobarServer is started and listening on 8066 16:59:19,527 INFO =============================================== 16:59:23,459 DEBUG 1>>NIOReactor接受连接数:0 16:59:23,464 DEBUG 2>>NIOReactor接受连接数:1 16:59:23,465 DEBUG select读事件 16:59:23,465 INFO com.alibaba.cobar.net.handler.FrontendAuthenticator接收的请求长度:62 58 0 0 1 5 166 15 0 0 0 0 1 33 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 114 111 111 116 0 20 169 171 247 102 133 96 158 224 121 22 226 229 88 244 119 238 185 61 124 219 16:59:23,468 INFO [thread=Processor1-H0,class=ServerConnection,host=192.168.137.8,port=46101,schema=null]'root' login success客户端得到的回复：
yan@yan-Z400:~$ mysql -uroot -p** -P8066 -h192.168.137.8 Welcome to the MySQL monitor. Commands end with ; or \g. Your MySQL connection id is 1 Server version: 5.1.48-cobar-1.2.7 Cobar Server (ALIBABA) Copyright (c) 2000, 2013, Oracle and/or its affiliates. All rights reserved. Oracle is a registered trademark of Oracle Corporation and/or its affiliates. Other names may be trademarks of their respective owners. Type 'help;' or '\h' for help. Type '\c' to clear the current input statement. mysql>MySQL客户端的命令处理，具体后续会分析。
本文出自：http://blog.geekcome.com, 原文地址：http://blog.geekcome.com/%e5%88%86%e5%b8%83%e5%bc%8f%e6%95%b0%e6%8d%ae%e5%ba%93%e4%b8%ad%e9%97%b4%e4%bb%b6-2-cobar%e4%b8%8e%e5%ae%a2%e6%88%b7%e7%ab%af%e7%9a%84%e6%8f%a1%e6%89%8b%e8%ae%a4%e8%af%81, 感谢原作者分享。
PHP实现开源SeaweedFS分布式文件系统Jun 18, 2023 pm 03:56 PM在分布式系统的架构中，文件管理和存储是非常重要的一部分。然而，传统的文件系统在应对大规模的文件存储和管理时遇到了一些问题。为了解决这些问题，SeaweedFS分布式文件系统被开发出来。在本文中，我们将介绍如何使用PHP来实现开源SeaweedFS分布式文件系统。什么是SeaweedFS？SeaweedFS是一个开源的分布式文件系统，它用于解决大规模文件存储和
Pandas 与 PySpark 强强联手，功能与速度齐飞！May 01, 2023 pm 09:19 PM使用Python做数据处理的数据科学家或数据从业者，对数据科学包pandas并不陌生，也不乏像云朵君一样的pandas重度使用者，项目开始写的第一行代码，大多是importpandasaspd。pandas做数据处理可以说是yyds！而他的缺点也是非常明显，pandas只能单机处理，它不能随数据量线性伸缩。例如，如果pandas试图读取的数据集大于一台机器的可用内存，则会因内存不足而失败。另外pandas在处理大型数据方面非常慢，虽然有像Dask或Vaex等其他库来优化提升数
PHP中的分布式数据中心May 23, 2023 pm 11:40 PM随着互联网的快速发展，网站的访问量也在不断增长。为了满足这一需求，我们需要构建高可用性的系统。分布式数据中心就是这样一个系统，它将各个数据中心的负载分散到不同的服务器上，增加系统的稳定性和可扩展性。在PHP开发中，我们也可以通过一些技术实现分布式数据中心。分布式缓存分布式缓存是互联网分布式应用中最常用的技术之一。它将数据缓存在多个节点上，提高数据的访问速度和
使用Redis实现分布式计数器May 11, 2023 am 08:06 AM什么是分布式计数器？在分布式系统中，多个节点之间需要对共同的状态进行更新和读取，而计数器是其中一种应用最广泛的状态之一。通俗地讲，计数器就是一个变量，每次被访问时其值就会加1或减1，用于跟踪某个系统进展的指标。而分布式计数器则指的是在分布式环境下对计数器进行操作和管理。为什么要使用Redis实现分布式计数器？随着分布式计算的普及，分布式系统中的许多细节问题也
分布式系统必须知道的一个共识算法：RaftApr 07, 2023 pm 05:54 PM一、Raft 概述Raft 算法是分布式系统开发首选的共识算法。比如现在流行 Etcd、Consul。如果掌握了这个算法，就可以较容易地处理绝大部分场景的容错和一致性需求。比如分布式配置系统、分布式 NoSQL 存储等等，轻松突破系统的单机限制。Raft 算法是通过一切以领导者为准的方式，实现一系列值的共识和各节点日志的一致。二、Raft 角色2.1 角色跟随者（Follower）：普通群众，默默接收和来自领导者的消息，当领导者心跳信息超时的
Redis实现分布式配置管理的方法与应用实例May 11, 2023 pm 04:22 PMRedis实现分布式配置管理的方法与应用实例随着业务的发展，配置管理对于一个系统而言变得越来越重要。一些通用的应用配置（如数据库连接信息，缓存配置等），以及一些需要动态控制的开关配置，都需要进行统一管理和更新。在传统架构中，通常是通过在每台服务器上通过单独的配置文件进行管理，但这种方式会导致配置文件的管理和同步变得十分复杂。因此，在分布式架构下，采用一个可靠
Redis实现分布式对象存储的方法与应用实例May 10, 2023 pm 08:48 PMRedis实现分布式对象存储的方法与应用实例随着互联网的快速发展和数据量的快速增长，传统的单机存储已经无法满足业务的需求，因此分布式存储成为了当前业界的热门话题。Redis是一个高性能的键值对数据库，它不仅支持丰富的数据结构，而且支持分布式存储，因此具有极高的应用价值。本文将介绍Redis实现分布式对象存储的方法，并结合应用实例进行说明。一、Redis实现分
PHP与数据库分布式的集成May 15, 2023 pm 09:40 PM随着互联网技术的发展，对于一个网络应用而言，对数据库的操作非常频繁。特别是对于动态网站，甚至有可能出现每秒数百次的数据库请求，当数据库处理能力不能满足需求时，我们可以考虑使用数据库分布式。而分布式数据库的实现离不开与编程语言的集成。PHP作为一门非常流行的编程语言，具有较好的适用性和灵活性，这篇文章将着重介绍PHP与数据库分布式集成的实践。分布式的概念分布式
熱AI工具
Undresser.AI Undress
人工智慧驅動的應用程序，用於創建逼真的裸體照片
AI Clothes Remover
用於從照片中去除衣服的線上人工智慧工具。
Undress AI Tool
免費脫衣圖片
Clothoff.io
AI脫衣器
AI Hentai Generator
免費產生 AI 無盡。
熱門文章
熱工具
SublimeText3漢化版
中文版，非常好用
MantisBT
Mantis是一個易於部署的基於Web的缺陷追蹤工具，用於幫助產品缺陷追蹤。它需要PHP、MySQL和一個Web伺服器。請查看我們的演示和託管服務。
MinGW - Minimalist GNU for Windows
這個專案正在遷移到osdn.net/projects/mingw的過程中，你可以繼續在那裡關注我們。 MinGW：GNU編譯器集合（GCC）的本機Windows移植版本，可自由分發的導入函式庫和用於建置本機Windows應用程式的頭檔；包括對MSVC執行時間的擴展，以支援C99功能。 MinGW的所有軟體都可以在64位元Windows平台上運作。
禪工作室 13.0.1
強大的PHP整合開發環境
SublimeText3 Mac版
神級程式碼編輯軟體(SublimeText3)
