自拍偷在线精品自拍偷,亚洲欧美中文日韩v在线观看不卡

C++多進(jìn)程并發(fā)框架FFLIB之Tutorial

作者：知然
開發(fā) 后端
FFLIB框架是為簡化分布式/多進(jìn)程并發(fā)而生的。它起始于本人嘗試解決工作中經(jīng)常遇到的問題如消息定義、異步、多線程、單元測試、性能優(yōu)化等。

 基本介紹可以看這里：

      http://www.cnblogs.com/zhiranok/archive/2012/07/30/fflib_framework.html

　　其中之所以特意采用了Broker模式，是吸收了MPI和Erlang的思想。

 關(guān)于MPI：http://www.mcs.anl.gov/research/projects/mpi/

 關(guān)于Erlang：http://www.erlang.org/

　　FFLIB 目前處于alpha階段，一些有用的功能還需繼續(xù)添加。但是FFLIB一開始就是為了解決實(shí)際問題而生。Broker 即可以以獨(dú)立進(jìn)程運(yùn)行，也可以集成到某個特定的進(jìn)程中啟動。除了這些，F(xiàn)FLIB中使用epoll實(shí)現(xiàn)的網(wǎng)絡(luò)層也***參考價值。網(wǎng)上有一些關(guān)于epoll ET 和 LT的討論，關(guān)于哪種方式更簡單，本人的答案是ET。ET模式下epoll 就是一個完全狀態(tài)機(jī)。開發(fā)者只需實(shí)現(xiàn)FD的read、write、error 三種狀態(tài)即可。

　　我進(jìn)一步挖掘FFLIB的功能。寫一篇FFLIB的Tutorial。創(chuàng)建更多的FFLIB使用示例，以此來深入探討FFLIB的意義。在游戲開發(fā)中，或者一些分布式的環(huán)境中，有許多大家熟悉的模式。，本文挑選了如下作為FFLIB示例：

Request/Reply

點(diǎn)對點(diǎn)通訊

阻塞通訊

多播通訊

Map/Reduce

Request/Reply

異步的Request/Reply

　　在FFLIB中所有的消息都是Request和Reply一一對應(yīng)的，默認(rèn)情況下工作在異步模式。假設(shè)如下場景，F(xiàn)lash連入GatewayServer并發(fā)送Login消息包，GatewaServer 解析用戶名密碼，調(diào)用LoginServer 驗(yàn)證。

首先定義msg:

  1. struct user_login_t 
  2. { 
  3.     struct in_t: public msg_i 
  4.     { 
  5.         in_t(): 
  6.             msg_i("user_login_t::in_t") 
  7.         {} 
  8.         string encode() 
  9.         { 
  10.             return (init_encoder() << uid << value).get_buff(); 
  11.         } 
  12.         void decode(const string& src_buff_) 
  13.         { 
  14.             init_decoder(src_buff_) >> uid >> value; 
  15.         } 
  16.         long   uid; 
  17.         string value; 
  18.     }; 
  19.     struct out_t: public msg_i 
  20.     { 
  21.         out_t(): 
  22.             msg_i("user_login_t::out_t") 
  23.         {} 
  24.         string encode() 
  25.         { 
  26.             return (init_encoder() << value).get_buff(); 
  27.         } 
  28.         void decode(const string& src_buff_) 
  29.         { 
  30.             init_decoder(src_buff_) >> value; 
  31.         } 
  32.         bool value; 
  33.     }; 
  34. }; 

LoginServer中如此定義接口：

  1. class login_server_t 
  2. { 
  3. public: 
  4.     void verify(user_login_t::in_t& in_msg_, rpc_callcack_t<user_login_t::out_t>& cb_) 
  5.     { 
  6.         user_login_t::out_t out; 
  7.         out.value = true; 
  8.         cb_(out); 
  9.     } 
  10. }; 
  11. login_server_t login_server; 
  12. singleton_t<msg_bus_t>::instance().create_service("login_server", 1) 
  13.             .bind_service(&login_server) 
  14.             .reg(&login_server_t::verify); 

在GatewayServer中調(diào)用上面接口：

  1. struct lambda_t 
  2.     { 
  3.         static void callback(user_login_t::out_t& msg_, socket_ptr_t socket_) 
  4.         { 
  5.             if (true == msg_.value) 
  6.             { 
  7.                 //! socket_->send_msg("login ok"); 
  8.             } 
  9.             else 
  10.             { 
  11.                 //! socket_->send_msg("login failed"); 
  12.             } 
  13.         } 
  14.     }; 
  15.     user_login_t::in_t in; 
  16.     in.uid  = 520; 
  17.     in.value = "ILoveYou"; 
  18.     socket_ptr_t flash_socket = NULL;//! TODO 
  19.     singleton_t<msg_bus_t>::instance() 
  20.          .get_service_group("login_server_t") 
  21.         ->get_service(1) 
  22.        ->async_call(in, binder_t::callback(&lambda_t::callback, flash_socket)); 

如上所示， async_call 可以通過binder_t模板函數(shù)為回調(diào)函綁定參數(shù)。

同步的Request/Reply

　　大部分時候我們期望Reply被異步處理，但有時Reply 必須被首先處理后才能觸發(fā)后續(xù)操作，一般這種情況發(fā)生在程序初始化之時。假設(shè)如下場景，SceneServer啟動時必須從SuperServer中獲取配置，然后才能執(zhí)行加載場景數(shù)據(jù)等后續(xù)初始化操作。

　　首先定義通信的msg：

  1. struct config_t 
  2. { 
  3.     struct in_t: public msg_i 
  4.     { 
  5.         in_t(): 
  6.             msg_i("config_t::in_t") 
  7.         {} 
  8.         string encode() 
  9.         { 
  10.             return (init_encoder() << server_type << server_id).get_buff(); 
  11.         } 
  12.         void decode(const string& src_buff_) 
  13.         { 
  14.             init_decoder(src_buff_) >> server_type >> server_id; 
  15.         } 
  16.         int server_type; 
  17.         int server_id; 
  18.     }; 
  19.     struct out_t: public msg_i 
  20.     { 
  21.         out_t(): 
  22.             msg_i("config_t::out_t") 
  23.         {} 
  24.         string encode() 
  25.         { 
  26.             return (init_encoder() << value).get_buff(); 
  27.         } 
  28.         void decode(const string& src_buff_) 
  29.         { 
  30.             init_decoder(src_buff_) >> value; 
  31.         } 
  32.         map<string, string> value; 
  33.     }; 
  34. }; 

如上所示， msg 序列化自動支持map。

　　SuperServer 中定義返回配置的接口：

  1. super_server_t super_server; 
  2. singleton_t<msg_bus_t>::instance().create_service("super_server", 1) 
  3.     .bind_service(&super_server) 
  4.     .reg(&super_server_t::get_config); 
  5. SceneServer 可以如此實(shí)現(xiàn)同步Request/Reply: 
  6. rpc_future_t<config_t::out_t> rpc_future; 
  7. config_t::in_t in; 
  8. in.server_type = 1; 
  9. in.server_id   = 1; 
  10. const config_t::out_t& out = rpc_future.call(  singleton_t<msg_bus_t>::instance().get_service_group("super_server") 
  11.         ->get_service(1), in); 
  12. cout << out.value.size() <<"\n"; 
  13. //std::foreach(out.value.begin(), out.value.end(), fuctor_xx); 

點(diǎn)對點(diǎn)通訊

　　異步Request/Reply 已經(jīng)能夠解決大部分問題了，但是有時處理Push模式時稍顯吃了。我們知道消息推算有Push 和Poll兩種方式。了解二者：

      http://blog.sina.com.cn/s/blog_6617106b0100hrm1.html

　　上面提到的Request/Reply 非常適合poll模式，以上一個獲取配置為例，SuperServer由于定義接口的時候只需知道callback，并不知道SceneServer的具體連接。，所以SuperServer不能向SceneServer Push消息。在FFLIB中并沒有限定某個節(jié)點(diǎn)必須是Client或只能是Service，實(shí)際上可以兼有二者的角色。SceneServer 也可以提供接口供SuperServer調(diào)用，這就符合了Push的語義。假設(shè)如下場景，GatewayServer需要在用戶登入時調(diào)用通知SessionServer，而某一時刻SessionServer也可能呢通知GatewayServer 強(qiáng)制某用戶下線。二者互為client和service。大家必須知道，在FFLIB中實(shí)現(xiàn)兩個節(jié)點(diǎn)的通信只需知道對方的服務(wù)名稱即可，Broker 在此時實(shí)現(xiàn)解耦的作用非常明顯，若要增加對其他節(jié)點(diǎn)的通信，只需通過服務(wù)名稱async_call即可。

　　定義通信的msg：

  1. struct user_online_t 
  2. { 
  3.     struct in_t: public msg_i 
  4.     { 
  5.         in_t(): 
  6.             msg_i("user_online_t::in_t") 
  7.         {} 
  8.         string encode() 
  9.         { 
  10.             return (init_encoder() << uid).get_buff(); 
  11.         } 
  12.         void decode(const string& src_buff_) 
  13.         { 
  14.             init_decoder(src_buff_) >> uid; 
  15.         } 
  16.         long uid; 
  17.     }; 
  18.     struct out_t: public msg_i 
  19.     { 
  20.         out_t(): 
  21.             msg_i("user_online_t::out_t") 
  22.         {} 
  23.         string encode() 
  24.         { 
  25.             return (init_encoder() << value).get_buff(); 
  26.         } 
  27.         void decode(const string& src_buff_) 
  28.         { 
  29.             init_decoder(src_buff_) >> value; 
  30.         } 
  31.         bool value; 
  32.     }; 
  33. }; 
  34. struct force_user_offline_t 
  35. { 
  36.     struct in_t: public msg_i 
  37.     { 
  38.         in_t(): 
  39.             msg_i("force_user_offline_t::in_t") 
  40.         {} 
  41.         string encode() 
  42.         { 
  43.             return (init_encoder() << uid).get_buff(); 
  44.         } 
  45.         void decode(const string& src_buff_) 
  46.         { 
  47.             init_decoder(src_buff_) >> uid; 
  48.         } 
  49.        long uid; 
  50.     }; 
  51.     struct out_t: public msg_i 
  52.     { 
  53.         out_t(): 
  54.             msg_i("force_user_offline_t::out_t") 
  55.         {} 
  56.        string encode() 
  57.         { 
  58.             return (init_encoder() << value).get_buff(); 
  59.         } 
  60.         void decode(const string& src_buff_) 
  61.         { 
  62.             init_decoder(src_buff_) >> value; 
  63.         } 
  64.         bool value; 
  65.     }; 
  66. }; 

GatewayServer 通知SessionServer 用戶上線，并提供強(qiáng)制用戶下線的接口：

  1. class gateway_server_t 
  2. { 
  3. public: 
  4.     void force_user_offline(force_user_offline_t::in_t& in_msg_, rpc_callcack_t<force_user_offline_t::out_t>& cb_) 
  5.     { 
  6.         //! close user socket 
  7.         force_user_offline_t::out_t out; 
  8.         out.value = true; 
  9.         cb_(out); 
  10.     } 
  11. }; 
  12. gateway_server_t gateway_server; 
  13. singleton_t<msg_bus_t>::instance().create_service("gateway_server", 1) 
  14.             .bind_service(&gateway_server) 
  15.             .reg(&gateway_server_t::force_user_offline); 
  16. user_online_t::in_t in; 
  17. in.uid = 520; 
  18. singleton_t<msg_bus_t>::instance() 
  19.     .get_service_group("session_server") 
  20.     ->get_service(1) 
  21.     ->async_call(in, callback_TODO); 

SessionServer 提供用戶上線接口，可能會調(diào)用GatewayServer 的接口強(qiáng)制用戶下線。

  1. class session_server_t 
  2. { 
  3. public: 
  4.     void user_login(user_online_t::in_t& in_msg_, rpc_callcack_t<user_online_t::out_t>& cb_) 
  5.     { 
  6.         //! close user socket 
  7.         user_online_t::out_t out; 
  8.         out.value = true; 
  9.         cb_(out); 
  10.     } 
  11. }; 
  12. session_server_t session_server; 
  13. singleton_t<msg_bus_t>::instance().create_service("session_server", 1) 
  14.             .bind_service(&session_server) 
  15.             .reg(&session_server_t::user_login); 
  16. force_user_offline_t::in_t in; 
  17. in.uid = 520; 
  18. singleton_t<msg_bus_t>::instance() 
  19.     .get_service_group("gateway_server") 
  20.     ->get_service(1) 
  21.     ->async_call(in, callback_TODO); 

多播通信

　　和點(diǎn)對點(diǎn)通信一樣，要實(shí)現(xiàn)多播，只需要知道目標(biāo)的服務(wù)名稱。特別提一點(diǎn)的是，F(xiàn)FLIB中有服務(wù)組的概念。比如啟動了多個場景服務(wù)器SceneServer，除了數(shù)據(jù)不同，二者接口完全相同，有可能只是相同進(jìn)程的不同實(shí)例。在FFLIB框架中把這些服務(wù)歸為一個服務(wù)組，然后再為每個實(shí)例分配索引id。

　　假設(shè)如下場景，SuperServer 中要實(shí)現(xiàn)一個GM接口，通知所有SceneServer 重新加載配置。

　　定義通信的msg：

  1. struct reload_config_t 
  2.  
  3.    struct in_t: public msg_i 
  4.    { 
  5.        in_t(): 
  6.            msg_i("reload_config_t::in_t") 
  7.        {} 
  8.        string encode() 
  9.        { 
  10.            return (init_encoder()).get_buff(); 
  11.        } 
  12.        void decode(const string& src_buff_) 
  13.        { 
  14.            init_decoder(src_buff_); 
  15.        } 
  16.   }; 
  17.    struct out_t: public msg_i 
  18.    { 
  19.       out_t(): 
  20.            msg_i("reload_config_t::out_t") 
  21.        {} 
  22.         string encode() 
  23.        { 
  24.            return (init_encoder() << value).get_buff(); 
  25.        } 
  26.        void decode(const string& src_buff_) 
  27.        { 
  28.            init_decoder(src_buff_) >> value; 
  29.        } 
  30.        bool value; 
  31.    }; 
  32. ; 

SceneServer 提供重新載入配置接口：

  1. class scene_server_t 
  2. { 
  3. public: 
  4.     void reload_config(reload_config_t::in_t& in_msg_, rpc_callcack_t<reload_config_t::out_t>& cb_) 
  5.     { 
  6.         //! close user socket 
  7.         reload_config_t::out_t out; 
  8.         out.value = true; 
  9.         cb_(out); 
  10.     } 
  11. }; 
  12. scene_server_t scene_server; 
  13. singleton_t<msg_bus_t>::instance().create_service("scene_server", 1) 
  14.             .bind_service(&scene_server) 
  15.             .reg(&scene_server_t::reload_config);  

在SuperServer 中如此實(shí)現(xiàn)多播（跟準(zhǔn)確是廣播，大同小異）：

  1. struct lambda_t 
  2. { 
  3. 　　static void reload_config(rpc_service_t* rs_) 
  4. 　　{ 
  5.           reload_config_t::in_t in; 
  6.           rs_->async_call(in, callback_TODO); 
  7. 　　} 
  8. }; 
  9. singleton_t<msg_bus_t>::instance() 
  10.     .get_service_group("scene_server") 
  11.     ->foreach(&lambda_t::reload_config); 

Map/Reduce

　　在游戲中使用Map/reduce 的情形并不多見，本人找到網(wǎng)上最常見的Map/reduce 實(shí)例 WordCount。情形如下：有一些文本字符串，統(tǒng)計(jì)每個字符出現(xiàn)的次數(shù)。

Map操作，將文本分為多個子文本，分發(fā)給多個Worker 進(jìn)程進(jìn)行統(tǒng)計(jì)

Reduce 操作，將多組worker 進(jìn)程計(jì)算的結(jié)果匯總

Worker：為文本統(tǒng)計(jì)各個字符出現(xiàn)的次數(shù)

定義通信消息： 

  1. struct word_count_t 
  2. { 
  3.     struct in_t: public msg_i 
  4.     { 
  5.         in_t(): 
  6.             msg_i("word_count_t::in_t") 
  7.         {} 
  8.         string encode() 
  9.         { 
  10.             return (init_encoder() << str).get_buff(); 
  11.         } 
  12.         void decode(const string& src_buff_) 
  13.         { 
  14.             init_decoder(src_buff_) >> str; 
  15.         } 
  16.         string str; 
  17.     }; 
  18.     struct out_t: public msg_i 
  19.     { 
  20.         out_t(): 
  21.             msg_i("word_count_t::out_t") 
  22.         {} 
  23.         string encode() 
  24.         { 
  25.             return (init_encoder() << value).get_buff(); 
  26.         } 
  27.         void decode(const string& src_buff_) 
  28.         { 
  29.             init_decoder(src_buff_) >> value; 
  30.         } 
  31.        map<char, int> value; 
  32.     }; 
  33.  
  34. }; 

定義woker的接口：

  1. class worker_t 
  2. { 
  3. public: 
  4.     void word_count(word_count_t::in_t& in_msg_, rpc_callcack_t<word_count_t::out_t>& cb_) 
  5.     { 
  6.         //! close user socket 
  7.         word_count_t::out_t out; 
  8.         for (size_t i = 0; i < in_msg_.str.size(); ++i) 
  9.         { 
  10.             map<int, int>::iterator it = out.value.find(in_msg_.str[i]); 
  11.             if (it != out.value.end()) 
  12.             { 
  13.                 it->second += 1; 
  14.             } 
  15.             else 
  16.             { 
  17.                 out.value[in_msg_.str[i]] = 1; 
  18.             } 
  19.         } 
  20.         cb_(out); 
  21.     } 
  22. }; 
  23. worker_t worker; 
  24.    for (int i = 0; i < 5; ++i) 
  25.     { 
  26.        singleton_t<msg_bus_t>::instance().create_service("worker", 1) 
  27.             .bind_service(&worker) 
  28.             .reg(&worker_t::word_count); 
  29.     } 

模擬Map/reduce 操作：

  1. struct lambda_t 
  2. { 
  3.     static void reduce(word_count_t::out_t& msg_, map<int, int>* result_, size_t* size_) 
  4.     { 
  5.         for (map<int, int>::iterator it = msg_.value.begin(); it != msg_.value.end(); ++it) 
  6.         { 
  7.             map<int, int>::iterator it2 = result_->find(it->first); 
  8.             if (it2 != result_->end()) 
  9.             { 
  10.                 it2->second += it->second; 
  11.             } 
  12.             else 
  13.             { 
  14.                 (*result_)[it->first] = it->second; 
  15.             } 
  16.         } 
  17.         if (-- size_ == 0) 
  18.         { 
  19.             //reduce end!!!!!!!!!!!!!!!! 
  20.             delete result_; 
  21.             delete size_; 
  22.         } 
  23.     } 
  24.     static void do_map(const char** p, size_t size_) 
  25.     { 
  26.         map<int, int>* result  = new map<int, int>(); 
  27.        size_t*    dest_size   = new size_t(); 
  28.         *dest_size = size_; 
  29.         for (size_t i = 0; i < size_; ++i) 
  30.         { 
  31.             word_count_t::in_t in; 
  32.             in.str = p[i]; 
  33.             singleton_t<msg_bus_t>::instance() 
  34.                 .get_service_group("worker") 
  35.                 ->get_service(1 + i % singleton_t<msg_bus_t>::instance().get_service_group("worker")->size()) 
  36.                ->async_call(in, binder_t::callback(&lambda_t::reduce, result, dest_size)); 
  37.         } 
  38.     } 
  39. }; 
  40. const char* str_vec[] = {"oh nice", "oh fuck", "oh no", "oh dear", "oh wonderful", "oh bingo"}; 
  41. lambda_t::do_map(str_vec, 6); 

總結(jié)：

FFLIB 使進(jìn)程間通信更容易

source code:  https://ffown.googlecode.com/svn/trunk

原文鏈接：http://www.cnblogs.com/zhiranok/archive/2012/08/08/fflib_tutorial.html

 

【編輯推薦】

 

責(zé)任編輯：彭凡 來源： 博客園
C++
相關(guān)推薦

2012-08-08 09:32:26

C++多進(jìn)程并發(fā)框架

2011-07-15 00:47:13

C++多態(tài)

2011-07-14 17:45:06

CC++

2017-06-30 10:12:46

Python多進(jìn)程

2011-07-10 15:26:54

C++

2011-07-13 18:24:18

C++

2022-04-01 13:10:20

C++服務(wù)器代碼

2024-12-27 08:11:44

Python編程模式IO

2023-11-22 12:25:05

C++RTTI

2015-04-21 13:37:44

Google開源CC++版

2020-07-30 12:40:35

CC++編程語言

2010-07-15 12:51:17

Perl多進(jìn)程

2024-09-29 10:39:14

并發(fā)Python多線程

2024-01-03 10:03:26

PythonTCP服務(wù)器

2010-02-01 10:54:37

C++框架

2023-12-13 10:51:49

C++函數(shù)模板編程

2023-11-28 11:51:01

C++函數(shù)

2024-02-01 00:10:21

C++PIMPL編程

2024-03-29 06:44:55

Python多進(jìn)程模塊工具

2016-01-11 10:29:36

Docker容器容器技術(shù)
點(diǎn)贊
收藏

51CTO技術(shù)棧公眾號