用戶自定義協議client/server代碼示例

代碼參考鏈接：https://github.com/sogou/workflow

message.h

message.cc

server.cc

client.cc

關于user_defined_protocol

本示例設計一個簡單的通信協議，并在協議上構建server和client。server將client發送的消息轉換成大寫并返回。

協議的格式

協議消息包含一個4字節的head和一個message
body。head是一個網絡序的整數，指明body的長度。

請求和響應消息的格式一致。

協議的實現

用戶自定義協議，需要提供協議的序列化和反序列化方法，這兩個方法都是ProtocolMeessage類的虛函數。

另外，為了使用方便，強烈建議用戶實現消息的移動構造和移動賦值（用于std::move()）。 在ProtocolMessage.h里，序列化反序列化接口如下：

namespace protocol

{

class ProtocolMessage

public CommMessageOut,
public CommMessageIn

{

private:

virtual

int encode(struct iovec
vectors[], int max);

/*

You have to implement one of the ‘append’ functions, and the first one

 * with

arguement ‘size_t *size’ is recommmended. */

virtual

int append(const void
*buf, size_t *size)；

virtual int append(const void

*buf, size_t size);

...

};

}

序列化函數encode

encode函數在消息被發送之前調用，每條消息只調用一次。
encode函數里，用戶需要將消息序列化到一個vector數組，數組元素個數不超過max。目前max的值為8192。
結構體struct iovec定義在請參考系統調用readv和writev。
encode函數正確情況下的返回值在0到max之間，表示消息使用了多少個vector。

如果是UDP協議，請注意總長度不超過64k，并且使用不超過1024個vector（Linux一次writev只能1024個vector）。

UDP協議只能用于client，無法實現UDP
server。

encode返回-1表示錯誤。返回-1時，需要置errno。如果返回值>max，將得到一個EOVERFLOW錯誤。錯誤都在callback里得到。
為了性能考慮vector里的iov_base指針指向的內容不會被復制。所以一般指向消息類的成員。

反序列化函數append

append函數在每次收到一個數據塊時被調用。因此，每條消息可能會調用多次。
buf和size分別是收到的數據塊內容和長度。用戶需要把數據內容復制。

如果實現了append(const void buf, size_t size)接口，可以通過修改size來告訴框架本次消費了多少長度。收到的size
- 消耗的size = 剩余的size，剩余的那部分buf會由下一次append被調用時再次收到。此功能更方便協議解析，當然用戶也可以全部復制自行管理，則無需修改size。

append函數返回0表示消息還不完整，傳輸繼續。返回1表示消息結束。-1表示錯誤，需要置errno。
總之append的作用就是用于告訴框架消息是否已經傳輸結束。不要在append里做復雜的非必要的協議解析。

errno的設置

encode或append返回-1或其它負數都會被理解為失敗，需要通過errno來傳遞錯誤原因。用戶會在callback里得到這個錯誤。
如果是系統調用或libc等庫函數失敗（比如malloc）,libc肯定會設置好errno，用戶無需再設置。
一些消息不合法的錯誤是比較常見的，比如可以用EBADMSG，EMSGSIZE分別表示消息內容錯誤，和消息太大。
用戶可以選擇超過系統定義errno范圍的值來表示一些自定義錯誤。一般大于256的值是可以用的。
請不要使用負數errno。因為框架內部用了負數來代表SSL錯誤。

示例里，消息的序列化反序列化都非常的簡單。

頭文件message.h里，聲明了request和response類：

namespace protocol

{

class TutorialMessage

public ProtocolMessage

{

private:

virtual

int encode(struct iovec
vectors[], int max);

virtual

int append(const void
*buf, size_t size);

...

};

using TutorialRequest = TutorialMessage;

using TutorialResponse = TutorialMessage;

}

request和response類，都是同一種類型的消息。直接using就可以。

注意request和response必須可以無參數的被構造，也就是說需要有無參數的構造函數，或完全沒有構造函數。

此外，通訊過程中，如果發生重試，response對象會被銷毀并重新構造。因此，它最好是一個RAII類。否則處理起來會比較復雜。

message.cc里包含了encode和append的實現：

namespace protocol

{

int TutorialMessage::encode(struct iovec
vectors[], int max/max==8192/)

{

uint32_t

n = htonl(this->body_size);

memcpy(this->head,

&n, 4);

vectors[0].iov_base = this->head;vectors[0].iov_len = 4;vectors[1].iov_base = this->body;vectors[1].iov_len = this->body_size;return

2; /* return the number of vectors used, no more then max.
*/

}

int TutorialMessage::append(const void
*buf, size_t size)

{

if

(this->head_received
< 4)

{size_t

head_left;

    void

*p;

    p = &this->head[this->head_received];head_left = 4 - this->head_received;if

(size < 4 - this->head_received)

    {memcpy(p,

buf, size);

        this->head_received += size;return

0;

    }memcpy(p,

buf, head_left);

    size -= head_left;buf = (const char

*)buf + head_left;

    p = this->head;this->body_size = ntohl(*(uint32_t *)p);if

(this->body_size

this->size_limit)

    {errno = EMSGSIZE;return

-1;

    }this->body = (char *)malloc(this->body_size);if

(!this->body)

        return

-1;

    this->body_received = 0;}size_t

body_left = this->body_size - this->body_received;

if

(size > body_left)

{errno = EBADMSG;return

-1;

}memcpy(this->body,

buf, body_left);

if

(size < body_left)

    return

0;

return

1;

}

}

encode的實現非常簡單，固定使用了兩個vector，分別指向head和body。需要注意iov_base指針必須指向消息類的成員。

append需要保證4字節的head接收完整，再讀取message body。而且我們并不能保證第一次append一定包含完整的head，所以過程略為繁瑣。

append實現了size_limit功能，超過size_limit的會返回EMSGSIZE錯誤。用戶如果不需要限制消息大小，可以忽略size_limit這個域。

由于要求通信協議是一來一回的，所謂的“TCP包”問題不需要考慮，直接當錯誤消息處理。

現在，有了消息的定義和實現，就可以建立server和client了。

server和client的定義

有了request和response類，我們就可以建立基于這個協議的server和client。前面的示例里介紹過Http協議相關的類型定義：

using WFHttpTask =
WFNetworkTask<protocol::HttpRequest,

                             protocol::HttpResponse>;

using http_callback_t
= std::function<void (WFHttpTask *)>;

using WFHttpServer =
WFServer<protocol::HttpRequest,

protocol::HttpResponse>;

using http_process_t
= std::function<void (WFHttpTask *)>;

同樣的，對這個Tutorial協議，數據類型的定義并沒有什么區別：

using WFTutorialTask =
WFNetworkTask<protocol::TutorialRequest,

protocol::TutorialResponse>;

using tutorial_callback_t
= std::function<void (WFTutorialTask
*)>;

using WFTutorialServer =
WFServer<protocol::TutorialRequest,

protocol::TutorialResponse>;

using tutorial_process_t
= std::function<void (WFTutorialTask
*)>;

server端

server與普通的http
server沒有什么區別。優先IPv6啟動，這不影響IPv4的client請求。另外限制請求最多不超過4KB。

代碼請自行參考server.cc

client端

client端的邏輯是從標準IO接收用戶輸入，構造出請求發往server并得到結果。

為了簡單，讀取標準輸入的過程都在callback里完成，因此我們會先發出一條空請求。同樣為了安全我們限制server回復包不超4KB。

client端唯一需要了解的就是怎么產生一個自定義協議的client任務，在WFTaskFactory.h有三個接口可以選擇：

template<class REQ, class RESP>

class WFNetworkTaskFactory

{

private:

using

T = WFNetworkTask<REQ, RESP>;

public:

static

T *create_client_task(TransportType type,

                             const std::string& host,unsigned short

port,

                             int retry_max,std::function<void (T *)> callback);static

T *create_client_task(TransportType type,

                             const std::string& url,int retry_max,

std::function<void (T *)> callback);

static

T *create_client_task(TransportType type,

                             const URI& uri,int retry_max,

std::function<void (T *)>
callback);

...

};

其中，TransportType指定傳輸層協議，目前可選的值包括TT_TCP，TT_UDP，TT_SCTP和TT_TCP_SSL。

三個接口的區別不大，在這個示例里暫時不需要URL，用域名和端口來創建任務。

實際的調用代碼如下。派生了WFTaskFactory類，但這個派生并非必須的。

using namespace
protocol;

class MyFactory

public WFTaskFactory

{

public:

static

WFTutorialTask *create_tutorial_task(const std::string& host,

unsigned short
port,

int retry_max,

tutorial_callback_t callback)

{using

NTF = WFNetworkTaskFactory<TutorialRequest, TutorialResponse>;

    WFTutorialTask *task = NTF::create_client_task(TT_TCP, host, port,

retry_max,

                                                   std::move(callback));task->set_keep_alive(30

• 1000);

    return
  

task;

}

};

可以看到用了WFNetworkTaskFactory<TutorialRequest,
TutorialResponse>類來創建client任務。

接下來通過任務的set_keep_alive()接口，讓連接在通信完成之后保持30秒，否則，將默認采用短連接。

client的其它代碼涉及的知識點在之前的示例里都包含了。請參考client.cc

內置協議的請求是怎么產生的

現在系統中內置了http,
redis，mysql，kafka四種協議。可以通過相同的方法產生一個http或redis任務嗎？比如：

WFHttpTask *task = WFNetworkTaskFactory<protocol::HttpRequest,
protocol::HttpResponse>::create_client_task(…);

需要說明的是，這樣產生的http任務，會損失很多的功能，比如，無法根據header來識別是否用持久連接，無法識別重定向等。

同樣，如果這樣產生一個MySQL任務，可能根本就無法運行起來。因為缺乏登錄認證過程。

一個kafka請求可能需要和多臺broker有復雜的交互過程，這樣創建的請求顯然也無法完成這一過程。

可見每一種內置協議消息的產生過程都遠遠比這個示例復雜。同樣，如果用戶需要實現一個更多功能的通信協議，還有許多代碼要寫。

總結

以上是生活随笔為你收集整理的用户自定义协议client/server代码示例的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。