1

server/bin/udpclient.js

@@ -0,0 +1,16 @@
+const dgram = require('dgram');
+const client = dgram.createSocket('udp4');
+
+client.on('close', () => {
+  console.log('close');
+});
+
+client.on('error', () => {
+  console.log('error');
+});
+
+client.on('message', (msg, rinfo) => {
+  console.log(`接收到来自：${rinfo.address}:${rinfo.port} 的消息： ${msg}`);
+});
+
+client.send('hello', 17601, '108.137.177.19');

server/bin/udpserver.js

@@ -0,0 +1,17 @@
+const dgram = require('dgram');
+
+const server = dgram.createSocket('udp4');
+
+server.on('message', (msg, rinfo) => {
+  console.log('rinfo.address = ' + rinfo.address);
+  console.log('rinfo.port = ' + rinfo.port);
+  console.log(msg.toString());
+  server.send(msg, rinfo.port, rinfo.address);
+});
+
+server.on('listening', () => {
+  console.log('address:' + server.address().address);
+  console.log('port:' + server.address().port);
+});
+
+server.bind(17601);

server/wsproxy/CMakeLists.txt

@@ -20,9 +20,9 @@ endif()
 
 set(CMAKE_CXX_FLAGS_RELEASE "-O3 -Wall -g -std=gnu++1z -DGAME_ID=${GAME_ID} -DNDEBUG")
 if (${ASAN})
-  set(CMAKE_CXX_FLAGS_DEBUG "-Wall -g -std=gnu++1z -DGAME_ID=${GAME_ID} -DDEBUG -fsanitize=address -fno-omit-frame-pointer")
+  set(CMAKE_CXX_FLAGS_DEBUG "-Wall -g -std=gnu++1z -DGAME_ID=${GAME_ID} -DDEBUG -fsanitize=address -fno-omit-frame-pointer -DUSE_ASIO=1")
 else()
-  set(CMAKE_CXX_FLAGS_DEBUG "-Wall -g -std=gnu++1z -DGAME_ID=${GAME_ID} -DDEBUG")
+  set(CMAKE_CXX_FLAGS_DEBUG "-Wall -g -std=gnu++1z -DGAME_ID=${GAME_ID} -DDEBUG -DUSE_ASIO=1")
 endif()
 
 include_directories(

server/wsproxy/app.cc

@@ -25,6 +25,7 @@
 #include "upstreammgr.h"
 #include "master.h"
 #include "mastermgr.h"
+#include "iomgr.h"
 
 #include "longsessionmgr.h"
 
@@ -127,6 +128,9 @@ bool App::Init(int argc, char* argv[])
     f8::Timer::Instance()->Init();
     JsonDataMgr::Instance()->Init();
     uuid_->SetMachineId((node_id_ - 1) * MAX_NODE_ID + instance_id_);
+    #ifdef USE_ASIO
+    IoMgr::Instance()->Init();
+    #endif
     DownStreamMgr::Instance()->Init();
     MasterMgr::Instance()->Init();
     UpStreamMgr::Instance()->Init();
@@ -189,6 +193,9 @@ void App::UnInit()
     MasterMgr::Instance()->UnInit();
     UpStreamMgr::Instance()->UnInit();
     DownStreamMgr::Instance()->UnInit();
+    #ifdef USE_ASIO
+    IoMgr::Instance()->UnInit();
+    #endif
     JsonDataMgr::Instance()->UnInit();
     f8::Timer::Instance()->UnInit();
     f8::MsgQueue::Instance()->UnInit();

server/wsproxy/constant.h

@@ -23,6 +23,13 @@ enum InnerMesssage_e
     IM_HandleRedirect
 };
 
+enum IoContext_e
+{
+   IC_Master,
+   IC_UpStream,
+   IC_Max
+};
+
 //网络处理对象
 enum NetHandler_e
 {

server/wsproxy/iomgr.cc

@@ -0,0 +1,41 @@
+#include "precompile.h"
+
+#include "iomgr.h"
+
+#ifdef USE_ASIO
+
+void IoMgr::Init()
+{
+    for (int i = 0; i < IC_Max; ++i) {
+        io_contexts_.push_back(std::vector<std::shared_ptr<asio::io_context>>());
+        auto& contexts = io_contexts_.at(i);
+        contexts.push_back(std::make_shared<asio::io_context>());
+        new std::thread(&IoMgr::WorkerThreadProc, this, contexts.at(0));
+    }
+}
+
+void IoMgr::UnInit()
+{
+
+}
+
+std::shared_ptr<asio::io_context> IoMgr::GetIoContext(int type)
+{
+    if (type >= 0 && type < io_contexts_.size()) {
+        auto& contexts = io_contexts_.at(type);
+        if (!contexts.empty()) {
+            return contexts.at(rand() % contexts.size());
+        }
+    }
+    return nullptr;
+}
+
+void IoMgr::WorkerThreadProc(std::shared_ptr<asio::io_context> io_context)
+{
+    while (true) {
+        io_context->run();
+        io_context->reset();
+    }
+}
+
+#endif

server/wsproxy/iomgr.h

@@ -0,0 +1,28 @@
+#pragma once
+
+#ifdef USE_ASIO
+
+#include <asio.hpp>
+
+class IoMgr : public a8::Singleton<IoMgr>
+{
+private:
+    IoMgr() {};
+    friend class a8::Singleton<IoMgr>;
+
+public:
+
+    void Init();
+    void UnInit();
+
+    std::shared_ptr<asio::io_context> GetIoContext(int type);
+
+ private:
+    void WorkerThreadProc(std::shared_ptr<asio::io_context> io_context);
+
+ private:
+
+    std::vector<std::vector<std::shared_ptr<asio::io_context>>> io_contexts_;
+};
+
+#endif

server/wsproxy/master.cc

@@ -2,7 +2,11 @@
 
 #include <string.h>
 
+#ifdef USE_ASIO
+#include <a8/asiotcpclient.h>
+#else
 #include <a8/tcpclient.h>
+#endif
 
 #include <f8/udplog.h>
 #include <f8/timer.h>
@@ -12,6 +16,7 @@
 #include "master.h"
 
 #include "app.h"
+#include "iomgr.h"
 
 const int PACK_MAX = 1024 * 64;
 
@@ -24,9 +29,15 @@ void Master::Init(int instance_id, const std::string& remote_ip, int remote_port
     recv_bufflen_ = 0;
     last_pong_tick = a8::XGetTickCount();
     recv_buff_ = (char*) malloc(PACK_MAX * 2);
-    tcp_client_ = std::make_shared<a8::TcpClient>();
+#ifdef USE_ASIO
-    tcp_client_->remote_address = remote_ip;
+    auto io_context = IoMgr::Instance()->GetIoContext(IC_Master);
-    tcp_client_->remote_port = remote_port;
+    if (!io_context) {
+        abort();
+    }
+    tcp_client_ = std::make_shared<a8::AsioTcpClient>(*io_context, remote_ip, remote_port);
+#else
+    tcp_client_ = std::make_shared<a8::TcpClient>(remote_ip, remote_port);
+#endif
     tcp_client_->on_error = std::bind(&Master::on_error, this, std::placeholders::_1, std::placeholders::_2);
     tcp_client_->on_connect = std::bind(&Master::on_connect, this, std::placeholders::_1);
     tcp_client_->on_disconnect = std::bind(&Master::on_disconnect, this, std::placeholders::_1);
@@ -72,29 +83,45 @@ bool Master::Connected()
     return tcp_client_->Connected();
 }
 
+#ifdef USE_ASIO
+void Master::on_error(a8::AsioTcpClient* sender, int errorId)
+#else
 void Master::on_error(a8::TcpClient* sender, int errorId)
+#endif
 {
     f8::UdpLog::Instance()->Error("Master errorid=%d remote_ip:%s remote_port:%d",
                                   {
                                    errorId,
-                                   sender->remote_address,
+                                   sender->GetRemoteAddress(),
-                                   sender->remote_port
+                                   sender->GetRemotePort()
                                   });
 }
 
+#ifdef USE_ASIO
+void Master::on_connect(a8::AsioTcpClient* sender)
+#else
 void Master::on_connect(a8::TcpClient* sender)
+#endif
 {
     recv_bufflen_ = 0;
     f8::UdpLog::Instance()->Info("masterserver connected", {});
 }
 
+#ifdef USE_ASIO
+void Master::on_disconnect(a8::AsioTcpClient* sender)
+#else
 void Master::on_disconnect(a8::TcpClient* sender)
+#endif
 {
     recv_bufflen_ = 0;
     f8::UdpLog::Instance()->Info("masterserver %d disconnected after 10s later reconnect", {instance_id});
 }
 
+#ifdef USE_ASIO
+void Master::on_socketread(a8::AsioTcpClient* sender, char* buf, unsigned int len)
+#else
 void Master::on_socketread(a8::TcpClient* sender, char* buf, unsigned int len)
+#endif
 {
     #if 0
     ++App::Instance()->perf.read_count;

server/wsproxy/master.h

@@ -5,6 +5,7 @@
 namespace a8
 {
     class TcpClient;
+    class AsioTcpClient;
 }
 
 class Master
@@ -31,17 +32,28 @@ class Master
     }
 
  private:
+#ifdef USE_ASIO
+    void on_error(a8::AsioTcpClient* sender, int errorId);
+    void on_connect(a8::AsioTcpClient* sender);
+    void on_disconnect(a8::AsioTcpClient* sender);
+    void on_socketread(a8::AsioTcpClient* sender, char* buf, unsigned int len);
+#else
     void on_error(a8::TcpClient* sender, int errorId);
     void on_connect(a8::TcpClient* sender);
     void on_disconnect(a8::TcpClient* sender);
     void on_socketread(a8::TcpClient* sender, char* buf, unsigned int len);
+#endif
 
     void CheckAlive();
 
  private:
     char *recv_buff_ = nullptr;
     unsigned int recv_bufflen_ = 0;
+#ifdef USE_ASIO
+    std::shared_ptr<a8::AsioTcpClient> tcp_client_;
+#else
     std::shared_ptr<a8::TcpClient> tcp_client_;
+#endif
     f8::TimerWp timer_wp_;
     f8::Attacher attacher_;
 };

server/wsproxy/upstream.cc

@@ -2,7 +2,11 @@
 
 #include <string.h>
 
+#ifdef USE_ASIO
+#include <a8/asiotcpclient.h>
+#else
 #include <a8/tcpclient.h>
+#endif
 
 #include <f8/udplog.h>
 #include <f8/timer.h>
@@ -13,6 +17,7 @@
 #include "upstream.h"
 
 #include "app.h"
+#include "iomgr.h"
 
 const int PACK_MAX = 1024 * 64 * 2;
 
@@ -28,9 +33,15 @@ void UpStream::Init(int instance_id, const std::string& remote_ip, int remote_po
     recv_bufflen_ = 0;
     last_pong_tick = a8::XGetTickCount();
     recv_buff_ = (char*) malloc(PACK_MAX * 2);
-    tcp_client_ = std::make_shared<a8::TcpClient>();
+#ifdef USE_ASIO
-    tcp_client_->remote_address = remote_ip;
+    auto io_context = IoMgr::Instance()->GetIoContext(IC_UpStream);
-    tcp_client_->remote_port = remote_port;
+    if (!io_context) {
+        abort();
+    }
+    tcp_client_ = std::make_shared<a8::AsioTcpClient>(*io_context, remote_ip, remote_port);
+#else
+    tcp_client_ = std::make_shared<a8::TcpClient>(remote_ip, remote_port);
+#endif
     tcp_client_->on_error = std::bind(&UpStream::on_error, this, std::placeholders::_1, std::placeholders::_2);
     tcp_client_->on_connect = std::bind(&UpStream::on_connect, this, std::placeholders::_1);
     tcp_client_->on_disconnect = std::bind(&UpStream::on_disconnect, this, std::placeholders::_1);
@@ -148,23 +159,31 @@ void UpStream::ForwardClientMsgEx(f8::MsgHdr* hdr)
     }
 }
 
+#ifdef USE_ASIO
+void UpStream::on_error(a8::AsioTcpClient* sender, int errorId)
+#else
 void UpStream::on_error(a8::TcpClient* sender, int errorId)
+#endif
 {
     f8::UdpLog::Instance()->Error("target server errorid=%d remote_ip:%s remote_port:%d",
                                   {
                                    errorId,
-                                   sender->remote_address,
+                                   sender->GetRemoteAddress(),
-                                   sender->remote_port
+                                   sender->GetRemotePort()
                                   });
 }
 
+#ifdef USE_ASIO
+void UpStream::on_connect(a8::AsioTcpClient* sender)
+#else
 void UpStream::on_connect(a8::TcpClient* sender)
+#endif
 {
     recv_bufflen_ = 0;
     f8::UdpLog::Instance()->Info("target server connected remote_ip:%s remote_port:%d",
                                  {
-                                   sender->remote_address,
+                                  sender->GetRemoteAddress(),
-                                   sender->remote_port
+                                  sender->GetRemotePort()
                                  });
     f8::MsgQueue::Instance()->PostMsg
         (
@@ -178,15 +197,19 @@ void UpStream::on_connect(a8::TcpClient* sender)
          );
 }
 
+#ifdef USE_ASIO
+void UpStream::on_disconnect(a8::AsioTcpClient* sender)
+#else
 void UpStream::on_disconnect(a8::TcpClient* sender)
+#endif
 {
     recv_bufflen_ = 0;
     f8::UdpLog::Instance()->Info("target server %d disconnected after 10s later reconnect "
                                  "remote_ip:%s remote_port:%d",
                                  {
                                   instance_id,
-                                  sender->remote_address,
+                                  sender->GetRemoteAddress(),
-                                  sender->remote_port
+                                  sender->GetRemotePort()
                                  });
     f8::MsgQueue::Instance()->PostMsg
         (
@@ -200,7 +223,11 @@ void UpStream::on_disconnect(a8::TcpClient* sender)
          );
 }
 
+#ifdef USE_ASIO
+void UpStream::on_socketread(a8::AsioTcpClient* sender, char* buf, unsigned int len)
+#else
 void UpStream::on_socketread(a8::TcpClient* sender, char* buf, unsigned int len)
+#endif
 {
     #if 0
     ++App::Instance()->perf.read_count;

server/wsproxy/upstream.h

@@ -5,6 +5,7 @@
 namespace a8
 {
     class TcpClient;
+    class AsioTcpClient;
 }
 
 struct UpStreamMsgNode
@@ -26,6 +27,7 @@ class UpStream
     a8::tick_t last_pong_tick = 0;
 
  public:
+
     void Init(int instance_id, const std::string& remote_ip, int remote_port);
     void UnInit();
 
@@ -54,10 +56,17 @@ class UpStream
     void ForwardClientMsgEx(f8::MsgHdr* hdr);
 
  private:
+#ifdef USE_ASIO
+    void on_error(a8::AsioTcpClient* sender, int errorId);
+    void on_connect(a8::AsioTcpClient* sender);
+    void on_disconnect(a8::AsioTcpClient* sender);
+    void on_socketread(a8::AsioTcpClient* sender, char* buf, unsigned int len);
+#else
     void on_error(a8::TcpClient* sender, int errorId);
     void on_connect(a8::TcpClient* sender);
     void on_disconnect(a8::TcpClient* sender);
     void on_socketread(a8::TcpClient* sender, char* buf, unsigned int len);
+#endif
 
     void CheckAlive();
     void AddStockMsg(unsigned short socket_handle, int msgid, ::google::protobuf::Message* msg,
@@ -66,7 +75,11 @@ class UpStream
  private:
     char *recv_buff_ = nullptr;
     unsigned int recv_bufflen_ = 0;
+#ifdef USE_ASIO
+    std::shared_ptr<a8::AsioTcpClient> tcp_client_;
+#else
     std::shared_ptr<a8::TcpClient> tcp_client_;
+#endif
     f8::TimerWp timer_wp_;
     f8::Attacher attacher_;
 

server/wsproxy/upstreammgr.cc

@@ -55,7 +55,6 @@ std::weak_ptr<UpStream> UpStreamMgr::RecreateUpStream(const std::string& host, i
     int instance_id = curr_id_;
     std::string remote_ip = host;
     int remote_port = port;
-
     std::shared_ptr<UpStream> conn = std::make_shared<UpStream>();
     conn->Init(instance_id, remote_ip, remote_port);
     id_hash_[conn->instance_id] = conn;

third_party/a8

@@ -1 +1 @@
-Subproject commit 1e577389c8a2870db9ddbf18577bfca24def049b
+Subproject commit 466448796dff77a16190519c25879cfe0bc534ec

third_party/f8

@@ -1 +1 @@
-Subproject commit 243bbe515ef4a01089f9a6cf608c93d4097018de
+Subproject commit 8a49aa65c8576cabe1672be55fc903a6b17e2c37