Swoole项目从 2012 年推出到现在已经有 5 年的历史，现在越来越多的互联网企业使用 Swoole来开发各类后台应用。受限于 PHP 的 ZendVM实现，PHP 程序无法使用多线程进行编程开发。应用程序中实现并行处理只能使用多进程模式。
做过多进程开发的 PHPer 都知道进程的内存隔离性。在程序中声明的 global全局数组，实际上并不是数据共享的，在一个进程内修改数组的值，在另外一个进程中是无效的。
$array = array();
function process1() {
global $array;
$array['test'] = 'hello world';
}
function process2() {
global $array;
//这里读取不到test的值
var_dump($array['test']);
}
这个进程隔离性给程序的开发带来的很多烦恼。比如实现一个聊天室程序，用户 A在进程1中处理，用户 B在进程2中处理， A和 B如果在同一个 group，这个 group在多线程环境中直接用 set表示， A和 B加到对应 group的 set中即可。但多进程环境中，用 PHP 的 array无法实现。一般可以有2个思路解决问题：
-
进程间通信，可以使用管道，向另外一个进程发送请求，获取数据的值
-
借助存储实现，如
Redis、MySQL、文件
这2个方案虽然可以实现，但都存在明显的缺点。方案一实现较为复杂，开发困难。方案二实现简单，但存在额外的 IO消耗，不是 纯内存操作，有性能瓶颈。基于 /dev/shm实现内存文件读写的方案，是一个不错的方案，但需要注意锁的操作，读写时需要额外的系统调用开销。
想要解决这个问题，必须实现一个基于共享内存的数据结构。在 PHP 中也有一些扩展模块可以使用。如 APCu、 Yac、 shm_put_var/shm_get_var
-
Yac：性能高，但由于底层实现的限制，无法保证一致性。只能作为Cache来使用 -
APCu：支持Key-Value式数据的读写，缺点是实现简单粗暴，锁的粒度太粗。高并发时存在大量锁的争抢，性能较差 -
shm系列函数：这个方案虽然能实现共享内存操作，但实际上底层实现非常简陋。一方面底层根本没有加锁，如果你要在并发环境中使用，需要自行实现锁的操作。另外，底层实际上是一个链表结构，数据较多时，查询性能非常差
swoole_table 介绍
为了解决多进程程序中数据共享的难题， Swoole扩展提供了 swoole_table数据结构。 Table的实现非常精巧，使用最方便，同时性能也是最好的。
$table = new swoole_table(1024);
$table->column('id', swoole_table::TYPE_INT, 4);
$table->column('name', swoole_table::TYPE_STRING, 64);
$table->column('num', swoole_table::TYPE_FLOAT);
$table->create();
$table->set('[email protected]', array('id' => 145, 'name' => 'rango', 'num' => 3.1415));
$table->set('[email protected]', array('id' => 358, 'name' => "Rango1234", 'num' => 3.1415));
$table->set('[email protected]', array('id' => 189, 'name' => 'rango3', 'num' => 3.1415));
$ret1 = $table->get('[email protected]');
$ret2 = $table->get('[email protected]');
$table->del('[email protected]');
Table实现了一个二维 Map结构，有点像 PHP 的二维数组，简单易用。在最新的 1.9.19中还可以使用 ArrayAccess接口以 array的方式操作 Table：
$table = new swoole_table(1024);
$table->column('id', swoole_table::TYPE_INT);
$table->column('name', swoole_table::TYPE_STRING, 64);
$table->column('num', swoole_table::TYPE_FLOAT);
$table->create();
$table['apple'] = array('id' => 145, 'name' => 'iPhone', 'num' => 3.1415);
$table['google'] = array('id' => 358, 'name' => "AlphaGo", 'num' => 3.1415);
$table['microsoft']['name'] = "Windows";
$table['microsoft']['num'] = '1997.03';
var_dump($table['apple']);
var_dump($table['microsoft']);
$table['google']['num'] = 500.90;
var_dump($table['google']);
Table 的优势
-
性能极高，全部是纯内存操作，没有任何系统调用和IO的开销。在
酷睿I5机器上测试，Table单进程单线程每秒可完成写操作300万次，读操作每秒可完成150万次。在24核服务器上，理论上每秒可实现数千万次读写操作。 -
使用数据行锁，底层使用了数据行锁自旋锁。多进程并发执行时，读写不同的
key不存在锁的争抢问题。只有同一CPU时间读写同一个Key才需要进行加锁操作。而且Table本身锁的粒度非常小，get、set操作内部只有少量内存读写的指令，可以在数百纳秒内完成操作。
Table 的局限性
-
Key最大长度不得超过64字节 -
必须在创建前规划好容量，一旦写满后，再
set新的数据会出现内存分配导致失败，无法实现动态扩容
因此使用 Table时尽可能地设置较大的内存尺寸，这样虽然会带来一定的内存浪费，但实际上现代服务器内存非常廉价，这个局限性在实际项目中的问题并不大。
swoole_table 实现原理
Table底层基于共享内存实现，所占内存取决于表格的尺寸 size、冲突率（默认 20%）、 column的设置（如上面的示例中每行需要 8+64+8字节）、 64字节 KEY的存储空间、管理结构的内存消耗。
Table 的内存申请
size_t row_num = table->size * (1 + table->conflict_proportion);
size_t row_memory_size = sizeof(swTableRow) + table->item_size;
size_t memory_size = row_num * row_memory_size;
memory_size += sizeof(swMemoryPool) + sizeof(swFixedPool) + ((row_num - table->size) * sizeof(swFixedPool_slice));
memory_size += table->size * sizeof(swTableRow *);
void *memory = sw_shm_malloc(memory_size);
swoole_table本身是一个 HashTable结构， Key会计算为 hash值，来散列到每一行。 HashTable结构会遇到 Hash冲突问题，两个完全不同的 Key可能计算的 hash值是同一个，这时需要使用链表来解决 Hash冲突。 Swoole底层会创建一个浮动的内存池 swFixedPool结构来管理这些冲突 Key的内存。默认会创建 size*20%数量的浮动内存池。在 1.9.19中可以自行定义冲突率。
$table = new swoole_table(65536, 0.9);
假如你的场景中 Hash冲突较多，可以调高冲突率，以申请一块较大的浮动内存池。
static swTableRow* swTable_hash(swTable *table, char *key, int keylen)
{
#ifdef SW_TABLE_USE_PHP_HASH
uint64_t hashv = swoole_hash_php(key, keylen);
#else
uint64_t hashv = swoole_hash_austin(key, keylen);
#endif
uint64_t index = hashv & table->mask;
assert(index < table->size);
return table->rows[index];
}
swTableRow* swTableRow_set(swTable *table, char *key, int keylen, swTableRow **rowlock)
{
if (keylen > SW_TABLE_KEY_SIZE)
{
keylen = SW_TABLE_KEY_SIZE;
}
swTableRow *row = swTable_hash(table, key, keylen);
*rowlock = row;
swTableRow_lock(row);
#ifdef SW_TABLE_DEBUG
int _conflict_level = 0;
#endif
if (row->active)
{
for (;;)
{
if (strncmp(row->key, key, keylen) == 0)
{
break;
}
else if (row->next == NULL)
{
table->lock.lock(&table->lock);
swTableRow *new_row = table->pool->alloc(table->pool, 0);
#ifdef SW_TABLE_DEBUG
conflict_count ++;
if (_conflict_level > conflict_max_level)
{
conflict_max_level = _conflict_level;
}
#endif
table->lock.unlock(&table->lock);
if (!new_row)
{
return NULL;
}
//add row_num
bzero(new_row, sizeof(swTableRow));
sw_atomic_fetch_add(&(table->row_num), 1);
row->next = new_row;
row = new_row;
break;
}
else
{
row = row->next;
#ifdef SW_TABLE_DEBUG
_conflict_level++;
#endif
}
}
}
else
{
#ifdef SW_TABLE_DEBUG
insert_count ++;
#endif
sw_atomic_fetch_add(&(table->row_num), 1);
}
memcpy(row->key, key, keylen);
row->active = 1;
return row;
}
-
使用
swTable_hash计算hash值，散列到对应的行 -
Key发生冲突时，需要调用table->pool->alloc从浮动内存池中分配内存 -
浮动内存池内存不足时，
alloc失败，这时无法写入数据到Table
数据自旋锁
当同一 CPU时间，多个进程同时读取某一行时，需要锁的争抢。
swTableRow_lock(row);
//内存操作
swTableRow_unlock(_rowlock);
swTableRow_lock 本身是一个自选锁，这里使用了 gcc编译器提供的 __sync_bool_compare_and_swap函数进行 CPU原子操作。多个进程同时读写某一行数据时，先得到锁的进程会执行内存读写操作，未得到锁的进程会进行 CPU自旋等待进程释放锁。
static sw_inline void sw_spinlock(sw_atomic_t *lock)
{
uint32_t i, n;
while (1)
{
if (*lock == 0 && sw_atomic_cmp_set(lock, 0, 1))
{
return;
}
if (SW_CPU_NUM > 1)
{
for (n = 1; n < SW_SPINLOCK_LOOP_N; n <<= 1)
{
for (i = 0; i < n; i++)
{
sw_atomic_cpu_pause();
}
if (*lock == 0 && sw_atomic_cmp_set(lock, 0, 1))
{
return;
}
}
}
swYield();
}
}
返回结果
使用 table::get方法时，从 Table共享内存中，读取数据写入到 PHP本地内存数组中。底层会根据列信息 table->columns，计算内存指针的偏移量，得到对应字段的值。
相关文章推荐
static inline void php_swoole_table_row2array(swTable *table, swTableRow *row, zval *return_value)
{
array_init(return_value);
swTableColumn *col = NULL;
swTable_string_length_t vlen = 0;
double dval = 0;
int64_t lval = 0;
char *k;
while(1)
{
col = swHashMap_each(table->columns, &k);
if (col == NULL)
{
break;
}
if (col->type == SW_TABLE_STRING)
{
memcpy(&vlen, row->data + col->index, sizeof(swTable_string_length_t));
sw_add_assoc_stringl_ex(return_value, col->name->str, col->name->length + 1, row->data + col->index + sizeof(swTable_string_length_t), vlen, 1);
}
else if (col->type == SW_TABLE_FLOAT)
{
memcpy(&dval, row->data + col->index, sizeof(dval));
sw_add_assoc_double_ex(return_value, col->name->str, col->name->length + 1, dval);
}
else
{
switch (col->type)
{
case SW_TABLE_INT8:
memcpy(&lval, row->data + col->index, 1);
sw_add_assoc_long_ex(return_value, col->name->str, col->name->length + 1, (int8_t) lval);
break;
case SW_TABLE_INT16:
memcpy(&lval, row->data + col->index, 2);
sw_add_assoc_long_ex(return_value, col->name->str, col->name->length + 1, (int16_t) lval);
break;
case SW_TABLE_INT32:
memcpy(&lval, row->data + col->index, 4);
sw_add_assoc_long_ex(return_value, col->name->str, col->name->length + 1, (int32_t) lval);
break;
default:
memcpy(&lval, row->data + col->index, 8);
sw_add_assoc_long_ex(return_value, col->name->str, col->name->length + 1, lval);
break;
}
}
}
}
文章评论