MySQL:插件回调的方式

作者简介：高鹏，笔名八怪。《深入理解MySQL主从原理》图书作者，同时运营个人公众号“MySQL学习”，持续分享遇到的有趣case以及代码解析！

一、核心类

• Observer_info:观察者

class Observer_info {  //插件观察者
public:
  void *observer; //这个void指针是具体的观察者，使用指针函数实现多态  
  st_plugin_int *plugin_int;  
  plugin_ref plugin  
  Observer_info(void *ob, st_plugin_int *p);
};

• 实际观察者 及void* 指向的对象，其中全部都是函数指针，这里通过函数指针指向了具体的函数，实现了插件的功能。其中的函数指针指向的实际函数就是需要用户自己实现的。

Trans_observer 结构体
Server_state_observer 结构体
Binlog_transmit_observer 结构体
Binlog_relay_IO_observer 结构体

实际上看具体实现的时候，搜索这些结构的名字，插件中如果实现会定义实际的函数名。如MGR中如下：

Trans_observer trans_observer = {
  sizeof(Trans_observer),  group_replication_trans_before_dml,  
  
  group_replication_trans_before_commit,  
  group_replication_trans_before_rollback,  
  group_replication_trans_after_commit,  
  group_replication_trans_after_rollback,};

• Delegate:委托者基类 其中包含

  Observer_info_list observer_info_list; //观察者链表，也就是Observer_info的一个链表  
  mysql_rwlock_t lock;//读写锁  
  MEM_ROOT memroot;//内存空间  
  bool inited;//是否初始化

并且实现了一些通用的函数，比如增加和删除插件

• 具体的委托者继承自Delegate

Trans_delegate ：事物相关 typedef Trans_observer Observer; 
Server_state_delegate ：服务器相关 typedef Server_state_observer Observer;
Binlog_transmit_delegate  ：传输相关 typedef Binlog_transmit_observer Observer;
Binlog_relay_IO_delegate  ：slave 相关typedef Binlog_relay_IO_observer Observer;

二、注册函数

举例rpl_handler.cc中

int register_trans_observer(Trans_observer *observer, void *p)
{
  return transaction_delegate->add_observer(observer, (st_plugin_int *)p);
}

observer已经初始化完成，注册即可。这里加入到了观察者队列。一旦加入这个链表则，在实际使用的时候就会遍历整个链表执行相应的函数。

三、重要的宏

• RUN_HOOK 宏 定义如下：

#define RUN_HOOK(group, hook, args)             \
  (group ##_delegate->is_empty() ?              \
     0 : group ##_delegate->hook args)
     
#define NO_HOOK(group) (group ##_delegate->is_empty())

这个宏会在MySQL中代码的相应合适的位置进行调用，进入插件定义的逻辑。

• FOREACH_OBSERVER 宏 定义如下：

#define FOREACH_OBSERVER(r, f, thd, args)                                                           \
  /*
       Use a struct to make sure that they are allocated adjacent, check     
       delete_dynamic().  
  */                                                                                                                                                \   
  Prealloced_array<plugin_ref, 8> plugins(PSI_NOT_INSTRUMENTED);        \ //定义一个插件数组  
  read_lock();                                                                                                                            \  
  Observer_info_iterator iter= observer_info_iter();                                                \ //迭代器  
  Observer_info *info= iter++;                                                                                           \ //  
  for (; info; info= iter++)                                                                                                      \  
  {                                                                                                                                                   \
      plugin_ref plugin=                                                                                                            \
            my_plugin_lock(0, &info->plugin);                                                                     \    
      if (!plugin)                                                                                                                            \    
      {                                                                                                                                               \
            /* plugin is not intialized or deleted, this is not an error */                       \      
            r= 0;                                                                                                                                  \      
            break;                                                                                                                               \    
      }                                                                                                                                                \    
      plugins.push_back(plugin);                                                                                           \    
      if (((Observer *)info->observer)->f                                                                           \
              && ((Observer *)info->observer)->f args)                                                      \    
      {                                                                                                                                                \
            r= 1;                                                                                                                                   \      
            sql_print_error("Run function '" #f "' in plugin '%s' failed",                          \
                                  info->plugin_int->name.str);                                                           \      
            break;                                                                                                                               \    
      }                                                                                                                                                \  
}                                                                                                                                                      \

实际上可以看到是在遍历相应的实际委托者的链表observer_info_list，执行相应的回表函数。

四、一个实际的列子

RUN_HOOK(transaction,
                 before_commit,                 
                 (thd, all,                  
                  thd_get_cache_mngr(thd)->get_binlog_cache_log(true),                  
                  thd_get_cache_mngr(thd)->get_binlog_cache_log(false),                  
                  max(max_binlog_cache_size,
                                                  max_binlog_stmt_cache_size))

根据RUN_HOOK定义 group ##_delegate->hook args 转换为：

transaction_delegate->before_commit(thd, all,
                  thd_get_cache_mngr(thd)->get_binlog_cache_log(true),                  
                  thd_get_cache_mngr(thd)->get_binlog_cache_log(false),                  
                  max(max_binlog_cache_size,
                                                  max_binlog_stmt_cache_size)

此处的transaction_delegate是一个已经初始化的并且已经有插件注册的Trans_delegate类的全局对象。因为Trans_delegate继承来自Delegate，而在Trans_delegate中实现了before_commit的逻辑。其中包含的宏调用

FOREACH_OBSERVER(ret, before_commit, thd, (¶m)); //这里会执行回调函数
宏定义：(#define FOREACH_OBSERVER(r, f, thd, args)  )

做回调，实际上他会遍历整个transaction_delegate中的观察者，这些观察者就是每一个插件实现的特定的GROUP的功能，所以FOREACH_OBSERVER宏的这一句

((Observer *)info->observer)->f args

就装换为了(Trans_observer *)info->observer)->before_commit(¶m) 其中info是一个Observer_info对象其中包含了一个VOID指针observer，可以转换为需要的类型，而Trans_observer是一个结构体其中全部都是函数指针before_commit是一个函数指针指向了group_replication_trans_before_commit，整个回调过程完成。

五、一张调用图

调用图如下：