08篇 要给Nacos的UDP通信功能点个赞
学习不用那么功利，二师兄带你从更高维度轻松阅读源码～
Nacos在服务注册功能中使用到了UDP的通信方式，主要功能就是用来辅助服务实例变化时对客户端进行通知。然而，对于大多数使用Nacos的程序员来说，可能还不知道这个功能，更别说灵活运用了。
看完整个源码的实现，还是要为这一功能点个赞的，可以说非常巧妙和实用。但在实现上有一些不足，文末会进行指出。
本篇文章就带大家从源码层面来分析一下Nacos 2.0中是如何基于UDP协议来实现服务实例变更的通知。
UDP通知基本原理
在分析源码之前，先来从整体上看一下Nacos中UDP的实现原理。
我们知道，UDP协议通信是双向的，没有所谓的客户端和服务端，因此在客户端和服务器端都会开启UDP的监听。客户端是单独开启一个线程来处理UDP消息的。当采用HTTP协议与注册中心通信时，，在客户端调用服务订阅接口时，会将客户端的UPD信息（IP和端口）上送到注册中心，注册中心以PushClient对象来进行封装和存储。
当注册中心有实例变化时，会发布一个ServiceChangeEvent事件，注册中心监听到这个事件之后，会遍历存储的PushClient，基于UDP协议对客户端进行通知。客户端接收到UDP通知，即可更新本地缓存的实例列表。
前面我们已经知道，基于HTTP协议进行服务注册时，会有一个实例更新的时间差，因为是通过客户端定时拉取服务器中的实例列表。如果拉取太频繁，注册中心压力比较大，如果拉取的周期比较长，实例的变化又没办法快速感知到。而UDP协议的通知，恰恰弥补了这一缺点，所以说，要为基于UDP通知这个功能点个赞。
下面就来看看源码层面是如何实现的。
客户端UDP通知监听与处理
客户端在实例化NamingHttpClientProxy时，在其构造方法中会初始化PushReceiver。
public NamingHttpClientProxy(String namespaceId, SecurityProxy securityProxy, ServerListManager serverListManager,
Properties properties, ServiceInfoHolder serviceInfoHolder) {
// ...
// 构建BeatReactor
this.beatReactor = new BeatReactor(this, properties);
// 构建UDP端口监听
this.pushReceiver = new PushReceiver(serviceInfoHolder);
// ...
}
PushReceiver的构造方法，如下：
public PushReceiver(ServiceInfoHolder serviceInfoHolder) {
try {
// 持有ServiceInfoHolder引用
this.serviceInfoHolder = serviceInfoHolder;
// 获取UDP端口
String udpPort = getPushReceiverUdpPort();
// 根据端口情况，构建DatagramSocket，如果未设置端口，则采用随机端口
if (StringUtils.isEmpty(udpPort)) {
this.udpSocket = new DatagramSocket();
} else {
this.udpSocket = new DatagramSocket(new InetSocketAddress(Integer.parseInt(udpPort)));
}
// 创建只有一个线程的ScheduledExecutorService
this.executorService = new ScheduledThreadPoolExecutor(1, new ThreadFactory() {
@Override
public Thread newThread(Runnable r) {
Thread thread = new Thread(r);
thread.setDaemon(true);
thread.setName("com.alibaba.nacos.naming.push.receiver");
return thread;
}
});
// 执行线程，PushReceiver实现了Runnable接口
this.executorService.execute(this);
} catch (Exception e) {
NAMING_LOGGER.error("[NA] init udp socket failed", e);
}
}
PushReceiver的构造方法做了以下操作：
- 第一、持有ServiceInfoHolder对象引用；
- 第二、获取UDP端口；
- 第三、实例化DatagramSocket对象，用于发送和接收Socket数据；
- 第四，创建线程池，并执行PushReceiver（实现了Runnable接口）；
既然PushReceiver实现了Runnable接口，run方法肯定是需要重新实现的：
@Override
public void run() {
while (!closed) {
try {
// byte[] is initialized with 0 full filled by default
byte[] buffer = new byte[UDP_MSS];
// 创建DatagramPacket用于存储接收到的报文
DatagramPacket packet = new DatagramPacket(buffer, buffer.length);
// 接收报文，在未接收到报文时会进行线程阻塞
udpSocket.receive(packet);
// 将报文转换为json格式
String json = new String(IoUtils.tryDecompress(packet.getData()), UTF_8).trim();
NAMING_LOGGER.info("received push data: " + json + " from " + packet.getAddress().toString());
// 将json格式的报文转换为PushPacket对象
PushPacket pushPacket = JacksonUtils.toObj(json, PushPacket.class);
String ack;
// 如果符合条件，则调用ServiceInfoHolder进行接收报文处理，并返回应答报文
if (PUSH_PACKAGE_TYPE_DOM.equals(pushPacket.type) || PUSH_PACKAGE_TYPE_SERVICE.equals(pushPacket.type)) {
serviceInfoHolder.processServiceInfo(pushPacket.data);
// send ack to server
ack = "{\"type\": \"push-ack\"" + ", \"lastRefTime\":\"" + pushPacket.lastRefTime + "\", \"data\":"
+ "\"\"}";
} else if (PUSH_PACKAGE_TYPE_DUMP.equals(pushPacket.type)) {
// dump data to server
ack = "{\"type\": \"dump-ack\"" + ", \"lastRefTime\": \"" + pushPacket.lastRefTime + "\", \"data\":"
+ "\"" + StringUtils.escapeJavaScript(JacksonUtils.toJson(serviceInfoHolder.getServiceInfoMap()))
+ "\"}";
} else {
// do nothing send ack only
ack = "{\"type\": \"unknown-ack\"" + ", \"lastRefTime\":\"" + pushPacket.lastRefTime
+ "\", \"data\":" + "\"\"}";
}
// 发送应答报文
udpSocket.send(new DatagramPacket(ack.getBytes(UTF_8), ack.getBytes(UTF_8).length,
packet.getSocketAddress()));
} catch (Exception e) {
if (closed) {
return;
}
NAMING_LOGGER.error("[NA] error while receiving push data", e);
}
}
}
PushReceiver#run方法主要处理了以下操作：
- 第一、构建DatagramPacket用于接收报文数据；
- 第二、通过DatagramSocket#receive方法阻塞等待报文的到来；
- 第三、DatagramSocket#receive接收到报文之后，方法继续执行；
- 第四、解析JSON格式的报文为PushPacket对象；
- 第五、判断报文类型，调用ServiceInfoHolder#processServiceInfo处理接收到的报文信息，在该方法中会将PushPacket转化为ServiceInfo对象；
- 第六、封装ACK信息（即应答报文信息）；
- 第七、通过DatagramSocket发送应答报文；
上面我们看到了Nacos客户端是如何基于UDP进行报文的监听和处理的，但并未找到客户端是如何将UDP信息上送给注册中心的。下面我们就来梳理一下，上送UDP信息的逻辑。
客户端上送UDP信息
在NamingHttpClientProxy中存储了UDP_PORT_PARAM，即UDP的端口参数信息。
UDP端口信息通过实例查询类接口进行传递，比如：查询实例列表、查询单个健康实例、查询所有实例、订阅接口、订阅的更新任务UpdateTask等接口。在这些方法中都调用了NamingClientProxy#queryInstancesOfService方法。
NamingHttpClientProxy中的queryInstancesOfService方法实现：
@Override
public ServiceInfo queryInstancesOfService(String serviceName, String groupName, String clusters, int udpPort,
boolean healthyOnly) throws NacosException {
final Map<String, String> params = new HashMap<String, String>(8);
params.put(CommonParams.NAMESPACE_ID, namespaceId);
params.put(CommonParams.SERVICE_NAME, NamingUtils.getGroupedName(serviceName, groupName));
params.put(CLUSTERS_PARAM, clusters);
// 获取UDP端口
params.put(UDP_PORT_PARAM, String.valueOf(udpPort));
params.put(CLIENT_IP_PARAM, NetUtils.localIP());
params.put(HEALTHY_ONLY_PARAM, String.valueOf(healthyOnly));
String result = reqApi(UtilAndComs.nacosUrlBase + "/instance/list", params, HttpMethod.GET);
if (StringUtils.isNotEmpty(result)) {
return JacksonUtils.toObj(result, ServiceInfo.class);
}
return new ServiceInfo(NamingUtils.getGroupedName(serviceName, groupName), clusters);
}
但查看源码会发现，查询实例列表、查询单个健康实例、查询所有实例、订阅的更新任务UpdateTask中，UDP端口传递的参数值均为0。只有HTTP协议的订阅接口取值为PushReceiver中的UDP端口号。
@Override
public ServiceInfo subscribe(String serviceName, String groupName, String clusters) throws NacosException {
return queryInstancesOfService(serviceName, groupName, clusters, pushReceiver.getUdpPort(), false);
}
在上面的代码中我们已经知道PushReceiver中有一个getPushReceiverUdpPort的方法：
public static String getPushReceiverUdpPort() {
return System.getenv(PropertyKeyConst.PUSH_RECEIVER_UDP_PORT);
}
很明显，UDP的端口是通过环境变量设置的，对应的key为“push.receiver.udp.port”。
而在1.4.2版本中，HostReactor中的NamingProxy成员变量的queryList方法也会传递UDP端口：
public void updateService(String serviceName, String clusters) throws NacosException {
ServiceInfo oldService = getServiceInfo0(serviceName, clusters);
try {
String result = serverProxy.queryList(serviceName, clusters, pushReceiver.getUdpPort(), false);
if (StringUtils.isNotEmpty(result)) {
processServiceJson(result);
}
} finally {
// ...
}
}
关于1.4.2版本中的实现，大家自行看源码即可，这里不再展开。
完成了客户端UDP基本信息的传递，再来看看服务器端是如何接收和存储这些信息的。
UDP服务存储
服务器端在获取实例列表的接口中，对UDP端口进行了处理。
@GetMapping("/list")
@Secured(parser = NamingResourceParser.class, action = ActionTypes.READ)
public Object list(HttpServletRequest request) throws Exception {
// ...
// 如果没有获得UDP端口信息，则默认端口为0
int udpPort = Integer.parseInt(WebUtils.optional(request, "udpPort", "0"));
// ...
// 客户端的IP、UDP端口封装到Subscriber对象中
Subscriber subscriber = new Subscriber(clientIP + ":" + udpPort, agent, app, clientIP, namespaceId, serviceName,
udpPort, clusters);
return getInstanceOperator().listInstance(namespaceId, serviceName, subscriber, clusters, healthyOnly);
}
在getInstanceOperator()方法中会获得当前采用的哪个协议，然后选择对应的处理类：
/**
* 判断并返回采用V1版本或V2版本的操作服务
* @return V1：Jraft协议（服务器端）；V2：gRpc协议（客户端）
*/
private InstanceOperator getInstanceOperator() {
return upgradeJudgement.isUseGrpcFeatures() ? instanceServiceV2 : instanceServiceV1;
}
这里具体的实现类为InstanceOperatorServiceImpl：
@Override
public ServiceInfo listInstance(String namespaceId, String serviceName, Subscriber subscriber, String cluster,
boolean healthOnly) throws Exception {
ClientInfo clientInfo = new ClientInfo(subscriber.getAgent());
String clientIP = subscriber.getIp();
ServiceInfo result = new ServiceInfo(serviceName, cluster);
Service service = serviceManager.getService(namespaceId, serviceName);
long cacheMillis = switchDomain.getDefaultCacheMillis();
// now try to enable the push
try {
// 处理支持UDP协议的客户端信息
if (subscriber.getPort() > 0 && pushService.canEnablePush(subscriber.getAgent())) {
subscriberServiceV1.addClient(namespaceId, serviceName, cluster, subscriber.getAgent(),
new InetSocketAddress(clientIP, subscriber.getPort()), pushDataSource, StringUtils.EMPTY,
StringUtils.EMPTY);
cacheMillis = switchDomain.getPushCacheMillis(serviceName);
}
} catch (Exception e) {
// ...
}
// ...
}
当UDP端口大于0，且agent参数定义的客户端支持UDP，则将对应的客户端信息封装到InetSocketAddress对象中，然后放入NamingSubscriberServiceV1Impl中（该类已经被废弃，看后续如何调整该方法实现）。
在NamingSubscriberServiceV1Impl中，会将对应的参数封装为PushClient，存放在Map当中。
public void addClient(String namespaceId, String serviceName, String clusters, String agent,
InetSocketAddress socketAddr, DataSource dataSource, String tenant, String app) {
PushClient client = new PushClient(namespaceId, serviceName, clusters, agent, socketAddr, dataSource, tenant,
app);
addClient(client);
}
addClient方法会将PushClient信息存放到ConcurrentMap\<String, ConcurrentMap\<String, PushClient>>当中:
private final ConcurrentMap<String, ConcurrentMap<String, PushClient>> clientMap = new ConcurrentHashMap<>();
public void addClient(PushClient client) {
// client is stored by key 'serviceName' because notify event is driven by serviceName change
String serviceKey = UtilsAndCommons.assembleFullServiceName(client.getNamespaceId(), client.getServiceName());
ConcurrentMap<String, PushClient> clients = clientMap.get(serviceKey);
if (clients == null) {
clientMap.putIfAbsent(serviceKey, new ConcurrentHashMap<>(1024));
clients = clientMap.get(serviceKey);
}
PushClient oldClient = clients.get(client.toString());
if (oldClient != null) {
oldClient.refresh();
} else {
PushClient res = clients.putIfAbsent(client.toString(), client);
// ...
}
}
此时，UDP的IP、端口信息已经封装到PushClient当中，并存储在NamingSubscriberServiceV1Impl的成员变量当中。
注册中心的UDP通知
当服务端发现某个实例发生了变化，比如主动注销了，会发布一个ServiceChangeEvent事件，UdpPushService会监听到该事件，并进行业务处理。
在UdpPushService的onApplicationEvent方法中，会根据PushClient的具体情况进行移除或发送UDP通知。onApplicationEvent中核心逻辑代码如下：
ConcurrentMap<String, PushClient> clients = subscriberServiceV1.getClientMap()
.get(UtilsAndCommons.assembleFullServiceName(namespaceId, serviceName));
if (MapUtils.isEmpty(clients)) {
return;
}
Map<String, Object> cache = new HashMap<>(16);
long lastRefTime = System.nanoTime();
for (PushClient client : clients.values()) {
// 移除僵尸客户端
if (client.zombie()) {
Loggers.PUSH.debug("client is zombie: " + client);
clients.remove(client.toString());
Loggers.PUSH.debug("client is zombie: " + client);
continue;
}
AckEntry ackEntry;
String key = getPushCacheKey(serviceName, client.getIp(), client.getAgent());
byte[] compressData = null;
Map<String, Object> data = null;
if (switchDomain.getDefaultPushCacheMillis() >= 20000 && cache.containsKey(key)) {
org.javatuples.Pair pair = (org.javatuples.Pair) cache.get(key);
compressData = (byte[]) (pair.getValue0());
data = (Map<String, Object>) pair.getValue1();
}
// 封装AckEntry对象
if (compressData != null) {
ackEntry = prepareAckEntry(client, compressData, data, lastRefTime);
} else {
ackEntry = prepareAckEntry(client, prepareHostsData(client), lastRefTime);
if (ackEntry != null) {
cache.put(key, new org.javatuples.Pair<>(ackEntry.getOrigin().getData(), ackEntry.getData()));
}
}
// 通过UDP通知其他客户端
udpPush(ackEntry);
}
事件处理的核心逻辑是就是先判断PushClient的状态信息，如果已经是僵尸客户端，则移除。然后将发送UDP的报文信息和接收客户端的信息封装为AckEntry对象，然后调用udpPush方法，进行UDP消息的发送。
注册中心的UDP接收
在看客户端源码的时候，我们看到客户端不仅会接收UDP请求，而且还会进行应答。那么注册中心怎么接收应答呢？也在UdpPushService类中，该类内部的静态代码块初始化一个UDP的DatagramSocket，用来接收消息：
static {
try {
udpSocket = new DatagramSocket();
Receiver receiver = new Receiver();
Thread inThread = new Thread(receiver);
inThread.setDaemon(true);
inThread.setName("com.alibaba.nacos.naming.push.receiver");
inThread.start();
} catch (SocketException e) {
Loggers.SRV_LOG.error("[NACOS-PUSH] failed to init push service");
}
}
Receiver是一个内部类，实现了Runnable接口，在其run方法中主要就是接收报文信息，然后进行报文消息的判断，根据判断结果，操作本地Map中数据。
UDP设计不足
文章最开始就写到，UDP的设计非常棒，即弥补了HTTP定时拉取的不足，又不至于太影响性能。但目前Nacos在UDP方面有一些不足，也可能是个人的吹毛求疵吧。
第一，文档中没有明确说明UDP的功能如何使用，这导致很多使用者在使用时并不知道UDP功能的存在，以及使用的限制条件。
第二，对云服务不友好。客户端的UDP端口可以自定义，但服务器端的UDP端口是随机获取到。在云服务中，即便是内网服务，UDP端口也是被防火墙限制的。如果服务端的UDP端口是随机获取（客户端默认也是），那么UDP的通信将直接被防火墙拦截掉，而用户根本看不到任何异常（UDP协议不关注客户端是否收到消息）。
至于这两点，说起来算是瑕不掩瑜，读完源码或读过我这篇文章的朋友大概已经知道怎么用了。后续可以给官方提一个Issue，看看是否可以改进。
小结
本文重点从三个方面讲解的Nacos基于UDP的服务实例变更通知：
第一，客户端监听UDP端口，当接收注册中心发来的服务实例变化，可以及时的更新本地的实例缓存；
第二，客户端通过订阅接口，将自身的UDP信息发送给注册中心，注册中心进行存储；
第三，注册中心中实例发生了变化，通过事件机制，将变更信息通过UDP协议发送给客户端。
经过本篇文章，想必你不仅了解了Nacos中UDP协议的通知机制。同时，也开拓了一个新的思路，即如何使用UDP，在什么场景下使用UDP，以及在云服务中使用UDP可能会存在的问题。如果这篇文章对你有帮助，关注或点赞都可以。
博主简介：《SpringBoot技术内幕》技术图书作者，酷爱钻研技术，写技术干货文章。
公众号：「程序新视界」，博主的公众号，欢迎关注~
技术交流：请联系博主微信号：zhuan2quan
关注公众号：程序新视界，一个让你软实力、硬技术同步提升的平台
除非注明，否则均为程序新视界原创文章，转载必须以链接形式标明本文链接
