一. 背景
最近的一个项目：我们的系统接收到上游系统的派单任务后，会推送到指定的门店的相关设备，并进行相应的业务处理。
二. Netty 的使用
在接收到派单任务之后，通过 Netty 推送到指定门店相关的设备。在我们的系统中 Netty 实现了消息推送、长连接以及心跳机制。
Netty+ZK.png
2.1 Netty Server 端:
每个 Netty 服务端通过 ConcurrentHashMap 保存了客户端的 clientId 以及它连接的 SocketChannel。
服务器端向客户端发送消息时，只要获取 clientId 对应的 SocketChannel，往 SocketChannel 里写入相应的 message 即可。
EventLoopGroup boss = new NioEventLoopGroup(1); EventLoopGroup worker = new NioEventLoopGroup(); ServerBootstrap bootstrap = new ServerBootstrap(); bootstrap.group(boss, worker) .channel(NioServerSocketChannel.class) .option(ChannelOption.SO_BACKLOG, 128) .option(ChannelOption.TCP_NODELAY, true) .childOption(ChannelOption.SO_KEEPALIVE, true) .childHandler(new ChannelInitializer() { @Override protected void initChannel(Channel channel) throws Exception { ChannelPipeline p = channel.pipeline(); p.addLast(new MessageEncoder()); p.addLast(new MessageDecoder()); p.addLast(new PushServerHandler()); } }); ChannelFuture future = bootstrap.bind(host,port).sync(); if (future.isSuccess()) { logger.info("server start..."); }
2.2 Netty Client 端:
客户端用于接收服务端的消息，随即进行业务处理。客户端还有心跳机制，它通过 IdleEvent 事件定时向服务端放送 Ping 消息以此来检测 SocketChannel 是否中断。
public PushClientBootstrap(String host, int port) throws InterruptedException { this.host = host; this.port = port; start(host,port); } private void start(String host, int port) throws InterruptedException { bootstrap = new Bootstrap(); bootstrap.channel(NioSocketChannel.class) .option(ChannelOption.SO_KEEPALIVE, true) .group(workGroup) .remoteAddress(host, port) .handler(new ChannelInitializer(){ @Override protected void initChannel(Channel channel) throws Exception { ChannelPipeline p = channel.pipeline(); p.addLast(new IdleStateHandler(20, 10, 0)); // IdleStateHandler 用于检测心跳 p.addLast(new MessageDecoder()); p.addLast(new MessageEncoder()); p.addLast(new PushClientHandler()); } }); doConnect(port, host); } /** * 建立连接，并且可以实现自动重连. * @param port port. * @param host host. * @throws InterruptedException InterruptedException. */ private void doConnect(int port, String host) throws InterruptedException { if (socketChannel != null && socketChannel.isActive()) { return; } final int portConnect = port; final String hostConnect = host; ChannelFuture future = bootstrap.connect(host, port); future.addListener(new ChannelFutureListener() { @Override public void operationComplete(ChannelFuture futureListener) throws Exception { if (futureListener.isSuccess()) { socketChannel = (SocketChannel) futureListener.channel(); logger.info("Connect to server successfully!"); } else { logger.info("Failed to connect to server, try connect after 10s"); futureListener.channel().eventLoop().schedule(new Runnable() { @Override public void run() { try { doConnect(portConnect, hostConnect); } catch (InterruptedException e) { e.printStackTrace(); } } }, 10, TimeUnit.SECONDS); } } }).sync(); }
三. 借助 ZooKeeper 实现简单的服务注册与发现
3.1 服务注册
服务注册本质上是为了解耦服务提供者和服务消费者。服务注册是一个高可用强一致性的服务发现存储仓库，主要用来存储服务的api和地址对应关系。为了高可用，服务注册中心一般为一个集群，并且能够保证分布式一致性。目前常用的有 ZooKeeper、Etcd 等等。
在我们项目中采用了 ZooKeeper 实现服务注册。
public class ServiceRegistry { private static final Logger logger = LoggerFactory.getLogger(ServiceRegistry.class); private CountDownLatch latch = new CountDownLatch(1); private String registryAddress; public ServiceRegistry(String registryAddress) { this.registryAddress = registryAddress; } public void register(String data) { if (data != null) { ZooKeeper zk = connectServer(); if (zk != null) { createNode(zk, data); } } } /** * 连接 zookeeper 服务器 * @return */ private ZooKeeper connectServer() { ZooKeeper zk = null; try { zk = new ZooKeeper(registryAddress, Constants.ZK_SESSION_TIMEOUT, new Watcher() { @Override public void process(WatchedEvent event) { if (event.getState() == Event.KeeperState.SyncConnected) { latch.countDown(); } } }); latch.await(); } catch (IOException | InterruptedException e) { logger.error("", e); } return zk; } /** * 创建节点 * @param zk * @param data */ private void createNode(ZooKeeper zk, String data) { try { byte[] bytes = data.getBytes(); String path = zk.create(Constants.ZK_DATA_PATH, bytes, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL); logger.debug("create zookeeper node ({} => {})", path, data); } catch (KeeperException | InterruptedException e) { logger.error("", e); } } }
有了服务注册，在 Netty 服务端启动之后，将 Netty 服务端的 ip 和 port 注册到 ZooKeeper。
EventLoopGroup boss = new NioEventLoopGroup(1); EventLoopGroup worker = new NioEventLoopGroup(); ServerBootstrap bootstrap = new ServerBootstrap(); bootstrap.group(boss, worker) .channel(NioServerSocketChannel.class) .option(ChannelOption.SO_BACKLOG, 128) .option(ChannelOption.TCP_NODELAY, true) .childOption(ChannelOption.SO_KEEPALIVE, true) .childHandler(new ChannelInitializer() { @Override protected void initChannel(Channel channel) throws Exception { ChannelPipeline p = channel.pipeline(); p.addLast(new MessageEncoder()); p.addLast(new MessageDecoder()); p.addLast(new PushServerHandler()); } }); ChannelFuture future = bootstrap.bind(host,port).sync(); if (future.isSuccess()) { logger.info("server start..."); } if (serviceRegistry != null) { serviceRegistry.register(host + ":" + port); }
3.2 服务发现
这里我们采用的是客户端的服务发现，即服务发现机制由客户端实现。
客户端在和服务端建立连接之前，通过查询注册中心的方式来获取服务端的地址。如果存在有多个 Netty 服务端的话，可以做服务的负载均衡。在我们的项目中只采用了简单的随机法进行负载。
public class ServiceDiscovery { private static final Logger logger = LoggerFactory.getLogger(ServiceDiscovery.class); private CountDownLatch latch = new CountDownLatch(1); private volatile List<String> serviceAddressList = new ArrayList<>(); private String registryAddress; // 注册中心的地址 public ServiceDiscovery(String registryAddress) { this.registryAddress = registryAddress; ZooKeeper zk = connectServer(); if (zk != null) { watchNode(zk); } } /** * 通过服务发现，获取服务提供方的地址 * @return */ public String discover() { String data = null; int size = serviceAddressList.size(); if (size > 0) { if (size == 1) { //只有一个服务提供方 data = serviceAddressList.get(0); logger.info("unique service address : {}", data); } else { //使用随机分配法。简单的负载均衡法 data = serviceAddressList.get(ThreadLocalRandom.current().nextInt(size)); logger.info("choose an address : {}", data); } } return data; } /** * 连接 zookeeper * @return */ private ZooKeeper connectServer() { ZooKeeper zk = null; try { zk = new ZooKeeper(registryAddress, Constants.ZK_SESSION_TIMEOUT, new Watcher() { @Override public void process(WatchedEvent event) { if (event.getState() == Watcher.Event.KeeperState.SyncConnected) { latch.countDown(); } } }); latch.await(); } catch (IOException | InterruptedException e) { logger.error("", e); } return zk; } /** * 获取服务地址列表 * @param zk */ private void watchNode(final ZooKeeper zk) { try { //获取子节点列表 List<String> nodeList = zk.getChildren(Constants.ZK_REGISTRY_PATH, new Watcher() { @Override public void process(WatchedEvent event) { if (event.getType() == Event.EventType.NodeChildrenChanged) { //发生子节点变化时再次调用此方法更新服务地址 watchNode(zk); } } }); List<String> dataList = new ArrayList<>(); for (String node : nodeList) { byte[] bytes = zk.getData(Constants.ZK_REGISTRY_PATH + "/" + node, false, null); dataList.add(new String(bytes)); } logger.debug("node data: {}", dataList); this.serviceAddressList = dataList; } catch (KeeperException | InterruptedException e) { logger.error("", e); } } }
Netty 客户端启动之后，通过服务发现获取 Netty 服务端的 ip 和 port。
/** * 支持通过服务发现来获取 Socket 服务端的 host、port * @param discoveryAddress * @throws InterruptedException */ public PushClientBootstrap(String discoveryAddress) throws InterruptedException { serviceDiscovery = new ServiceDiscovery(discoveryAddress); serverAddress = serviceDiscovery.discover(); if (serverAddress!=null) { String[] array = serverAddress.split(":"); if (array!=null && array.length==2) { String host = array[0]; int port = Integer.parseInt(array[1]); start(host,port); } } }
四. 总结
服务注册和发现一直是分布式的核心组件。本文介绍了借助 ZooKeeper 做注册中心，如何实现一个简单的服务注册和发现。其实，注册中心的选择有很多，例如 Etcd、Eureka 等等。选择符合我们业务需求的才是最重要的。
