nginx-push-stream模块源码学习（三）——发布

一、概述 
    发布：发布者将MSG post到某一特定通道上，channel将信息缓存 
在说明发布流程之前有必要说明下channel和msg的数据结构。 
二、数据结构 
2.1 MSG 
    发布时，模块先将消息转化为ngx_http_push_stream_msg_t的数据结构进行存储。 

Java代码  收藏代码
  1. // message queue  
  2. typedef struct {  
  3.     ngx_queue_t                     queue; // this MUST be first  
  4.     time_t                          expires;//消息过期时间  
  5.     time_t                          time;//消息创建时间  
  6.     ngx_flag_t                      deleted;//是否已删除  
  7.     ngx_int_t                       id;  
  8.     ngx_str_t                      *raw;//纯文本  
  9.     ngx_int_t                       tag;  
  10.     ngx_str_t                      *event_id;//支持event source  
  11.     ngx_str_t                      *event_id_message;  
  12.     ngx_str_t                      *formatted_messages;//格式化后消息  
  13.     ngx_int_t                       workers_ref_count;//待发送该消息ngx worker计数  
  14. } ngx_http_push_stream_msg_t;  


    @queue：每个channel会维护一个消息链表，每向channel发布一条消息，channel将其添加到自身的消息链表中。 
    @expires，@deleted：消息的过期时间，待介绍过订阅流程后，我会整理出一条消息的生产周期，到时会详细阐述该字段的意义。 
    @raw，@formatted_messages:该模块允许自定义三种消息模板——header模板：当收到订阅请求后发送模板消息；message模板：对消息体格式化；footer模板：断开连接时发送该模板。raw为消息原始内容，后者为应用message模板格式化后的信息 

2.2 channel 
    channel作为发布订阅的中间载体，理解理解它的存储至关重要。 

C代码  收藏代码
  1. typedef struct {  
  2.     ngx_rbtree_node_t                   node; // this MUST be first  
  3.     ngx_str_t                           id;  
  4.     ngx_uint_t                          last_message_id;  
  5.     time_t                              last_message_time;  
  6.     ngx_int_t                           last_message_tag;  
  7.     ngx_uint_t                          stored_messages;//# of messages  
  8.     ngx_uint_t                          subscribers;//# of subscribers  
  9.     ngx_http_push_stream_pid_queue_t    workers_with_subscribers;//处理该channel上订阅者的ngx worker进程链表  
  10.     ngx_http_push_stream_msg_t          message_queue;//消息链表  
  11.     time_t                              expires;//过期时间  
  12.     ngx_flag_t                          deleted;//是否已删除  
  13.     ngx_flag_t                          broadcast;//是否为广播通道  
  14.     ngx_http_push_stream_msg_t         *channel_deleted_message;//已删除消息链表  
  15. } ngx_http_push_stream_channel_t;  


2.3 worker msg 

Java代码  收藏代码
  1. // messages to worker processes  
  2. typedef struct {  
  3.     ngx_queue_t                         queue;  
  4.     ngx_http_push_stream_msg_t         *msg; // ->shared memory  
  5.     ngx_pid_t                           pid;  
  6.     ngx_http_push_stream_channel_t     *channel; // ->shared memory  
  7.     ngx_http_push_stream_queue_elem_t  *subscribers_sentinel; // ->a worker's local pool  
  8. } ngx_http_push_stream_worker_msg_t;  

  
    模块初始化时为每个ngx worker分配一片独立的工作区，工作区中维护一份消息链表。 
三、流程 
发布流程总的流程图如图所示： 
 
    对于删除channel和获取channel info的流程比较简单，不做阐述，具体说明下发布消息流程，流程图如图所示： 

 


    需要说明的是“向所有订阅者发送MSG”的过程： 

  • 向每个有该channel订阅者的worker（workers_with_subscriber）的消息链表中插入一条消息
  • 向上述worker发送CHECK_MESSAGES指令，触发msg发送流程（ngx_http_push_stream_process_worker_message）


MSG发送（ngx_http_push_stream_process_worker_message）： 

Java代码  收藏代码
  1.         // now let's respond to some requests!  
  2.         //对于该channel上的所有订阅者  
  3.         while ((cur = (ngx_http_push_stream_queue_elem_t *) ngx_queue_next(&cur->queue)) != subscribers_sentinel) {  
  4.             ngx_http_push_stream_subscriber_t *subscriber = (ngx_http_push_stream_subscriber_t *) cur->value;  
  5.             //如果订阅者为longpolling模式  
  6.             if (subscriber->longpolling) {  
  7.                 ngx_http_push_stream_queue_elem_t *prev = (ngx_http_push_stream_queue_elem_t *) ngx_queue_prev(&cur-  
  8. >queue);  
  9.                 //发送longpolling头（last Modified/Etag）  
  10.                 ngx_http_push_stream_add_polling_headers(subscriber->request, msg->time, msg->tag, subscriber->reque  
  11. st->pool);  
  12.                 ngx_http_send_header(subscriber->request);  
  13.                 //发送模块配置header模板  
  14.                 ngx_http_push_stream_send_response_content_header(subscriber->request, ngx_http_get_module_loc_conf(  
  15. subscriber->request, ngx_http_push_stream_module));  
  16.                 //发送响应MSG  
  17.                 ngx_http_push_stream_send_response_message(subscriber->request, channel, msg);  
  18.                 //发送footer模板，last chunck（""CRLF CRLF）  
  19.                 ngx_http_push_stream_send_response_finalize(subscriber->request);  
  20.   
  21.                 cur = prev;  
  22.             } else {//stream或polling模式  
  23.                 if (ngx_http_push_stream_send_response_message(subscriber->request, channel, msg) == NGX_ERROR) {  
  24.                     ngx_http_push_stream_queue_elem_t *prev = (ngx_http_push_stream_queue_elem_t *) ngx_queue_prev(&  
  25. cur->queue);  
  26.                     ngx_http_push_stream_send_response_finalize(subscriber->request);  
  27.                     cur = prev;  
  28.                 }  
  29.             }  


说明： 
    可以看出push stream模块在发布过程中针对longpolling和stream两种模式的不同： 

    • Longpolling模式下，每次发布消息时会发送longpolling头：last modified和etag，使得客户端下次请求时可据此判断服务端是否有更新的消息待发布。
    • longpolling模式下，订阅者每次请求都会在获得数据后断开重连，所以每次发布时都会发送header模板
    • ngx_http_push_stream_send_response_finalize同时会清理订阅者
【推广】 免费学中医，健康全家人
原文地址：https://www.cnblogs.com/xinyaoxp/p/3423547.html