Netty+WebSocket 获取火币交易所数据项目

<center>Netty+WebSocket 获取火币交易所时时数据项目</center> ==============================================

先附上项目项目GitHub地址 spring-boot-netty-websocket-huobi

项目简介

本项目使用 SpringBoot+Netty来开发WebSocket服务器,与火币交易所Websocket建立连接,时时获取火币网交易所推送过来的交易对最新数据。

该项目可以直接运用于实际开发中，做为获取各大交易所最新交易对相关数据的项目。

项目本身也是我在之前公司为了获取各大交易所数据所开发的项目,现在只是重新整理了下代码，现在它更像一个脚手架项目，可以在此基础上很方便的添加其它交易所。

技术架构

SpringBoot2.1.5 +Netty4.1.25 + Maven3.5.4 + lombok(插件)

项目测试

直接启动Springboot启动类Application.java,就可以时时获取火币网推送过来交易对的数据了。

如图

一、项目概述

1、项目启动入口

在项目启动的时候就开始去连接火币交易所Websocket订阅数据。

1   /**
2     * 首次启动并订阅火币websocket数据
3     */
4    @PostConstruct
5    public void firstSub() {
6        try {
7            huobiProMainService.start();
8        } catch (Exception e) {
9            log.error("huobi 首次启动订阅异常", e);
10        }
11    }

2、获取交易对数据

我们是先要获取火币交易所所有的交易对数据，然后告诉火币交易所我需要订阅哪些交易对数据。

是订阅所有交易对数据还是订阅部分交易对数据。

1    @Override
2    public synchronized List<String> getChannelCache() {
3        // 假设这里是从远处拉取交易对数据
4        List<String> list = Lists.newArrayList("btcusdt");
5        return list;
6    }

3、连接火币交易所Websocket,并订阅指定的交易对。

先与火币网WebSocket建立连接,连接成功后再告诉它我要订阅哪些交易对，哪种主题，成功后，火币交易所就会根据我们所订阅的主题和交易对，给我们时时推送消息。

1 /**
2     * 首次订阅交易对数据
3     *
4     * @param channelList 交易对列表
5     * @param topicFormat 交易对订阅主题格式
6     */
7    private void firstSub(List<String> channelList, String topicFormat) {
8        //封装huoBiProWebSocketService对象
9        klineClient = new HuoBiProWebSocketClient(huoBiProWebSocketService);
10        //启动连接火币网websocket
11        klineClient.start();
12        for (String channel : channelList) {
13            //订阅具体交易对
14            klineClient.addSub(formatChannel(topicFormat, channel));
15        }
16    }

启动连接火币网websocket核心代码

很明显我们我们是作为客户端去获取服务端的数据，所以这里的Bootstrap来与服务端进行数据交互，而不是用ServerBootstrap。

还有一点就是作为客户端，我们是要获取服务端所推送来的消息，所以我们自定义的handler是入站Handler,所以这里选择的是SimpleChannelInboundHandler。

1   /**
2         * 连接WebSocket，
3         *
4         * @param uri url构造出URI
5         * @param handler 处理消息
6         */
7        protected void connectWebSocket(final URI uri, SimpleChannelInboundHandler handler) {
8            try {
9                String scheme = uri.getScheme() == null ? "http" : uri.getScheme();
10                final String host = uri.getHost() == null ? "127.0.0.1" : uri.getHost();
11                final int port;
12
13                if (uri.getPort() == -1) {
14                    if ("http".equalsIgnoreCase(scheme) || "ws".equalsIgnoreCase(scheme)) {
15                        port = 80;
16                    } else if ("wss".equalsIgnoreCase(scheme)) {
17                        port = 443;
18                    } else {
19                        port = -1;
20                    }
21                } else {
22                    port = uri.getPort();
23                }
24
25                if (!"ws".equalsIgnoreCase(scheme) && !"wss".equalsIgnoreCase(scheme)) {
26                    System.out.println("Only WS(S) is supported");
27                    throw new UnsupportedAddressTypeException();
28                }
29                final boolean ssl = "wss".equalsIgnoreCase(scheme);
30                final SslContext sslCtx;
31                if (ssl) {
32                    sslCtx = SslContextBuilder.forClient().trustManager(InsecureTrustManagerFactory.INSTANCE).build();
33                } else {
34                    sslCtx = null;
35                }
36
37                group = new NioEventLoopGroup(2);
38                //构建客户端Bootstrap
39                Bootstrap bootstrap = new Bootstrap();
40                bootstrap.group(group).channel(NioSocketChannel.class).handler(new ChannelInitializer<SocketChannel>() {
41                    @Override
42                    protected void initChannel(SocketChannel ch) throws Exception {
43                        ChannelPipeline pipeline = ch.pipeline();
44                        if (sslCtx != null) {
45                            pipeline.addLast(sslCtx.newHandler(ch.alloc(), host, port));
46                        }
47                        //pipeline可以同时放入多个handler,最后一个为自定义hanler
48                        pipeline.addLast(new HttpClientCodec(), new HttpObjectAggregator(8192), handler);
49                    }
50                });
51                channel = bootstrap.connect(host, port).sync().channel();
52            } catch (Exception e) {
53                log.error(" webSocketClient start error.", e);
54                if (group != null) {
55                    group.shutdownGracefully();
56                }
57            }
58    }

4、自定义handler

自定义Handler才是核心，作为数据的入站这里选择继承SimpleChannelInboundHandler，继承它必须要实现一个方法就是channelRead0，通过该方法的msg，就可以获取火币交易所时时推送过来的消息了。

1/**
2 * @Description: 火币网WebSocket 消息处理类
3 * 自定义入站的handler 这个也是核心类
4 */
5@Slf4j
6public class HuoBiProWebSocketClientHandler extends SimpleChannelInboundHandler<Object> {
7
8    private WebSocketClientHandshaker handshaker;
9    private HuoBiProWebSocketClient client;
10    
11    /**
12     * 该handel获取消息的方法
13     */
14    @Override
15    protected void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception {
16        Channel channel = ctx.channel();
17        WebSocketFrame frame = (WebSocketFrame) msg;
18        if (frame instanceof BinaryWebSocketFrame) {
19            //火币网的数据是压缩过的，所以需要我们进行解压
20            BinaryWebSocketFrame binaryFrame = (BinaryWebSocketFrame) frame;
21            //获取数据、保存数据
22            client.onReceive(decodeByteBuf(binaryFrame.content()));
23        } else if (frame instanceof TextWebSocketFrame) {
24            TextWebSocketFrame textWebSocketFrame = (TextWebSocketFrame) frame;
25            client.onReceive(textWebSocketFrame.text());
26        } 
27     }
28  }

二、项目注意点

1、服务器问题

一般交易所的服务器都在国外，所以我们本地是无法建立Websocket连接的，除非本地翻墙。

同样项目也不能部署到阿里云等国内服务器，你只能选择香港或者国外服务器部署项目。

这里是火币网专门为我们提供的国内测试地址，所以本地可以获取数据。

2、获取交易所最新交易对数据问题

我们在向交易所Websocket订阅交易对的时候，首先就是要知道该交易所有哪些交易对，这份数据是需要我们单独去获取的，而且不是一次获取就好了。

因为该交易所可能新增或者删除交易对。所以需要我们通过定时任务去获取更新最新的交易对数据。

我这边只是模拟了一个交易对btcusdt,并没有提供获取最新交易对数据的服务。

3、数据存储问题

这也是最值得思考的一个问题，数据我们是获取了，但如果保存！

正常合理的开发应该获取数据是一个微服务，处理获取的数据是一个微服务。那么只需要获取数据后去调处理数据微服务就可以保存数据了。

但在这里，如果只是这样是行不通的。

因为火币网向我们推送的消息的速度会比我们调其它服务保存的数据要快，这就会存在数据丢失的情况发生。

这里仅仅是输出一个btcusdt交易对,并且只是订阅一个k线主题,而实际上交易所会有上百个交易对和几种订阅主题,

这样的消息推送速度是上面的几百倍。所以你会发现如果你不做任何改动，对于一些大的交易所而言，你的数据是来不及存储的。 补充

这边之前也写过有关 Netty 和 Websocket 相关的博客文章，可以做个参考

1、Netty专题（共9篇）

2、Websocket专题（共5篇）

 我相信，无论今后的道路多么坎坷，只要抓住今天，迟早会在奋斗中尝到人生的甘甜。抓住人生中的一分一秒，胜过虚度中的一月一年！(1）