ijk中的消息处理逻辑

作者: nullyy | 来源:发表于2017-01-05 19:18 被阅读305次

ijk中的消息处理逻辑
WebSocketHandlerDecorator的使用方法
String Format
mac系统 ijk（Android）编译
从demo分析ijk源码三：IjkMediaPlayer的jni
比特币源码解读六
RabbitMQ面试题：如何处理重复消息？（即消息幂等性保证）
mq 面试题
音视频、聊天
W7L5 W7L6

IJK中的消息处理

原因

ijk的给我们暴露了几个通知，让我们可以方便的使用通知来监控播放器的状态，不过我们总是觉得通知有各种各样的问题和局限性，并习惯性的把通知当成最后没有办法的选择。所以我深入ijk的底层研究了下它的消息机制，希望提供一套高效及时安全的播放器状态监听机制；

ijk状态监听原理

我们在ijkplayer.c 中发现了ijkmp_prepare_async_l方法；
其中使用

 mp->msg_thread = SDL_CreateThreadEx(&mp->_msg_thread, ijkmp_msg_loop, mp, "ff_msg_loop");

方法创建了一个线程，并且执行了ijkmp_msg_loop方法，ijkmp_msg_loop方式是在我们初始化ijkplayer是调用的ijkmp_ios_create方法时传入的函数指针：

 _mediaPlayer = ijkmp_ios_create(media_player_msg_loop);

其中media_player_msg_loop方法的实现如下：

       // 消息处理函数
       int media_player_msg_loop(void* arg)
     {
         @autoreleasepool {
             IjkMediaPlayer *mp = (IjkMediaPlayer*)arg;
             __weak livePlayer *livePlayer = ffplayerRetain(ijkmp_set_weak_thiz(mp, NULL));
             // 消息线程中 通过这个死循环，无限读取消息
             while (livePlayer) {
                 @autoreleasepool {
                     FFMoviePlayerMessage *msg = [livePlayer obtainMessage];
                     if (!msg)
                         break;
                     
                     int retval = ijkmp_get_msg(mp, &msg->_msg, 1);
                     if (retval < 0)
                         break;
                     
                     // block-get should never return 0
                     assert(retval > 0);
                     [livePlayer performSelectorOnMainThread:@selector(postEvent:) withObject:msg waitUntilDone:NO];
                 }
             }
             
             // retained in prepare_async, before SDL_CreateThreadEx
             ijkmp_dec_ref_p(&mp);
             return 0;
         }
     }

该函数主要做了三个事情:

创建了一个while死循环用于循环读取消息
通过ijkmp_get_msg从消息链表获取消息

通过postEvent方法将消息发送出去

其中ijkmp_get_msg函数实现了无消息时消息线程的阻塞和有新消息时取消线程阻塞的操作，函数实现如下

int ijkmp_get_msg(IjkMediaPlayer *mp, AVMessage *msg, int block)
  {
      assert(mp);
      while (1) {
          int continue_wait_next_msg = 0;
          // msg_queue_get只会在有新消息来是才会有返回值 不然它的线程阻塞的
          int retval = msg_queue_get(&mp->ffplayer->msg_queue, msg, block);
          if (retval <= 0)
              return retval;
  
          switch (msg->what) {
          .....
          }
  
          if (continue_wait_next_msg)
              continue;
  
          return retval;
      }
  
      return -1;
  }

该方法中通过msg_queue_get函数获取到消息链表内的消息,然后判断消息类型做相应的操作；
其中msg_queue_get方法是实际从消息链表取消息和开启和阻塞线程关键：

  // 整个函数的目的是: 从链表中取消息，从第一个开始取，取完将第二个设置成第一个，当没有消息的时候调用SDL_CondWait阻塞消息线程，等待新的消息加入
  inline static int msg_queue_get(MessageQueue *q, AVMessage *msg, int block)
  {
      AVMessage *msg1;
      int ret;
  
      SDL_LockMutex(q->mutex);
  
      for (;;) {
          if (q->abort_request) {
              ret = -1;
              break;
          }
  
          msg1 = q->first_msg;
          if (msg1) {
              // 这个步骤主要目的是 取出第一个msg
              q->first_msg = msg1->next;
              if (!q->first_msg)
                  q->last_msg = NULL;
              q->nb_messages--;
              *msg = *msg1;
  #ifdef FFP_MERGE
              av_free(msg1);
  #else
              msg1->next = q->recycle_msg;
              q->recycle_msg = msg1;
  #endif
              ret = 1;
              break;
          } else if (!block) {
              ret = 0;
              break;
          } else {
              // 没有msg 阻塞住线程
              // 添加条件锁 是线程阻塞的 只有在接到pthread_cond_signal才会打开走下面逻辑
              SDL_CondWait(q->cond, q->mutex);
          }
      }
      SDL_UnlockMutex(q->mutex);
      return ret;
  }

通过一个for死循环遍历消息链表中的消息(后面会详细分析消息链表),并且当消息全部读完后通过SDL_CondWait(q->cond, q->mutex)函数将消息线程再次阻塞掉，等待下次新的消息进来，如何激活消息线程会在下文提到；

接下来我们分析下ijk中的消息链表,首先我们分析下AVMessage结构体
```
 typedef struct AVMessage {
     int what;// 消息类型
     int arg1;
     int arg2;
     struct AVMessage *next;// 指向下个消息的指针
 } AVMessage;
```
消息结构体中有一个next指针，用于指向另一个AVMessage结构体，这个就是最常见的链表实现机制了；

我们再看一下加入新消息的实现：

我们在ijk代码中可以看到很多地方调用了ijkmp_change_state_l函数，这个函数就是改变播放器状态的接口了，当然还有许多其他加入消息的接口，这里就不一一列出了。我们可以看到该函数的调用堆栈

ijkmp_change_state_l -> ffp_notify_msg1 -> msg_queue_put_simple3 -> msg_queue_put -> msg_queue_put_private 终于我们来到了最终加入消息的方法

     // 推送消息到消息队列
 inline static int msg_queue_put_private(MessageQueue *q, AVMessage *msg)
 {
     AVMessage *msg1;
 
     if (q->abort_request)
         return -1;
 
 #ifdef FFP_MERGE
     msg1 = av_malloc(sizeof(AVMessage));
 #else
     msg1 = q->recycle_msg;
     if (msg1) {
         // 有循环消息 将循环消息指针指向当前循环消息后一消息,去除当前的循环消息
         q->recycle_msg = msg1->next;
         q->recycle_count++;
     } else {
         // 分配内存空间
         q->alloc_count++;
         msg1 = av_malloc(sizeof(AVMessage));
     }
 #ifdef FFP_SHOW_MSG_RECYCLE
     int total_count = q->recycle_count + q->alloc_count;
     if (!(total_count % 10)) {
         av_log(NULL, AV_LOG_DEBUG, "msg-recycle \t%d + \t%d = \t%d\n", q->recycle_count, q->alloc_count, total_count);
     }
 #endif
 #endif
     if (!msg1)
         return -1;
 
     // 将msg指针指向的 消息结构体赋值给msg1 是结构体的赋值
     *msg1 = *msg;
     msg1->next = NULL;
 
     if (!q->last_msg)
         q->first_msg = msg1;// 没有消息时，将传入的消息设置成第一个消息
     else
         q->last_msg->next = msg1;// 将新消息加入链表最后
     q->last_msg = msg1;
     q->nb_messages++;
     SDL_CondSignal(q->cond);
     return 0;
 }

该函数主要作用就是将新的消息放入消息链表的最后面，再调动SDL_CondSignal(q->cond)函数，激活消息线程，使msg_queue_get方法继续执行，最终最上层的media_player_msg_loop函数内部死循环得以继续执行通过postEvent方法抛出消息，等到消息全部抛出后，再继续阻塞消息线程；