分享IT技术，分享生活感悟，热爱摄影，热爱航天。
之前尝试做异地间的Memcache数据同步，使用Gearman作为消息队列配合消息处理程序进行数据写入，发现在内网中测试的性能和通过异地之间的专线性能差距非常明显，在异地的情况下即使增加大量的消息处理程序同步的数据流量依然难以增加。由此判断可能是由于异地之间专线的延迟还是要远远高于内网，读写Memcache这种要求低延迟的IO操作会严重受到网络延迟的影响（如果Memcache支持批量写入或许能够缓解这一问题）。因此想到的解决办法就是避免使用高延迟网络做读写Memcache的操作，即将数据先通同步到另一端，然后在进行写入就可以避免网络延迟带来的影响。
而对于将数据同步到另一端，可以通过一些数据库的复制机制进行，一般会比较容易想到Redis，将数据案时间分List写入，另一端使用脚本循环利用BLPOP方法读取数据进行同步。这也是个很好的办法，难点在于需要自行开发同步数据的程序，程序需要有较好的性能和稳定性，同时还需要便于维护和监控。
然而说到数据复制机制，我首先想到的就是MySQL，使用Blackhole Engine将数据通过binlog的方式同步到另一端，可靠性要高于Redis——复制中断可恢复，数据也有binlog作为记录也是不错的。对于进行同步的程序有两种办法，一种是通过程序读取复制过来的binlog，这个方法的难点依然在于需要自行开发同步程序，难点与使用Redis进行同步类似，另外一种方法是使用触发器进行同步，每接收一条INSERT语句执行一条对应的Memcache操作即可，这里还需要用到MySQL libmemcached UDF。
MySQL的UDF安装非常简单，首先下载[libmemcached UDF源代码](https://launchpad.net/memcached-udfs https://launchpad.net/memcached-udfs)，其依赖libmemcached，并且不能使用太新的版本，Centos6中使用的0.31是可以的。安装过程如下 ./configure make make install cp /usr/local/lib/libmemcached_functions_mysql.so /usr/lib64/mysql/plugin/ mysql -u root < sql/install_functions.sql
为了便于测试和最后的部署可能都需要用到多MySQL实例，这里记录一下操作的方法，首先初始化两个MySQL数据目录
mysql_install_db --user=mysql --datadir=/var/lib/mysql/3306/ mysql_install_db --user=mysql --datadir=/var/lib/mysql/3307/然后使用类似以下的配置文件，由于只需要使用Blackhole Engine，大多参数也无太大意义，设置小一些反而节约一些资源
[mysqld_multi] mysqld = /usr/bin/mysqld_safe mysqladmin = /usr/bin/mysqladmin [mysqld3306] port = 3306 socket = /var/lib/mysql/3306/mysqld.sock datadir = /var/lib/mysql/3306 key_buffer_size = 16M max_allowed_packet = 1M table_open_cache = 64 sort_buffer_size = 512K net_buffer_length = 8K read_buffer_size = 256K read_rnd_buffer_size = 512K myisam_sort_buffer_size = 8M log-bin=mysql-bin binlog_format=mixed server-id = 3306 innodb_data_file_path = ibdata1:10M:autoextend innodb_buffer_pool_size = 16M innodb_additional_mem_pool_size = 2M innodb_log_file_size = 5M innodb_log_buffer_size = 8M innodb_flush_log_at_trx_commit = 2 innodb_lock_wait_timeout = 50 [mysqld3307] port = 3307 socket = /var/lib/mysql/3307/mysqld.sock datadir = /var/lib/mysql/3307 key_buffer_size = 16M max_allowed_packet = 1M table_open_cache = 64 sort_buffer_size = 512K net_buffer_length = 8K read_buffer_size = 256K read_rnd_buffer_size = 512K myisam_sort_buffer_size = 8M log-bin=mysql-bin binlog_format=mixed server-id = 3307 innodb_data_file_path = ibdata1:10M:autoextend innodb_buffer_pool_size = 16M innodb_additional_mem_pool_size = 2M innodb_log_file_size = 5M innodb_log_buffer_size = 8M innodb_flush_log_at_trx_commit = 2 innodb_lock_wait_timeout = 50启动多实例的方法如下，启动两个实例
mysqld_multi start 3306 mysqld_multi start 3307
在MySQL启动后需要手动初始化libmemcached UDF，设置对应的Memcache服务器，同时设置分布式的散列方式，初始化脚本如下
SELECT memc_servers_set('127.0.0.1:11211,127.0.0.1:11212'); SELECT memc_servers_behavior_set('MEMCACHED_BEHAVIOR_DISTRIBUTION', 'MEMCACHED_DISTRIBUTION_CONSISTENT_KETAMA'); SELECT memc_servers_behavior_set('MEMCACHED_BEHAVIOR_KETAMA_WEIGHTED', 1);注意，每次MySQL启动时都需要进行以上的初始化，如下测试一下是否初始化成功，返回1则说明写入成功
SELECT memc_set('abc', '123'); +------------------------+ | memc_set('abc', '123') | +------------------------+ | 1 | +------------------------+ SELECT memc_get('abc'); +-----------------+ | memc_get('abc') | +-----------------+ | 123 | +-----------------+
libmemcached UDF有一个很大的缺点，就是Memcache的flags参数写死为0，而我们在同步的过程中需要将原来的flags值也进行同步，因此需要对libmemcached UDF的代码进行一点的调整，这里只改了set方法，其他的方法修改方式都是类似的。
diff -Nur memcached_functions_mysql-1.1/src/args.c memcached_functions_mysql-1.1.patched/src/args.c --- memcached_functions_mysql-1.1/src/args.c 2009-11-11 10:56:52.000000000 +0800 +++ memcached_functions_mysql-1.1.patched/src/args.c 2015-09-19 10:11:59.062512226 +0800 @@ -120,6 +120,30 @@ case MEMC_ADD_BY_KEY: case MEMC_REPLACE: case MEMC_REPLACE_BY_KEY: + { + + if (args->arg_count >= max_args - 1) + { + if (args->arg_type[max_args - 2] == STRING_RESULT) + { + container->flags= (uint32_t)atoi(args->args[max_args - 2]); + } + else if (args->arg_type[max_args - 2] == INT_RESULT) + { + container->flags= *((uint32_t*)args->args[max_args -2]); + } + else + { + container->flags= (uint32_t)0; + } + + } + else + { + container->flags= (uint32_t) 0; + } + fprintf(stderr, "flags %d\n", (int) container->expiration); + } case MEMC_APPEND: case MEMC_APPEND_BY_KEY: case MEMC_PREPEND: diff -Nur memcached_functions_mysql-1.1/src/common.h memcached_functions_mysql-1.1.patched/src/common.h --- memcached_functions_mysql-1.1/src/common.h 2009-11-15 06:53:23.000000000 +0800 +++ memcached_functions_mysql-1.1.patched/src/common.h 2015-09-19 09:50:30.480512194 +0800 @@ -11,6 +11,7 @@ struct memc_function_st { unsigned int offset; time_t expiration; + uint32_t flags; memcached_st memc; memcached_result_st results; memcached_string_st *stats_string; diff -Nur memcached_functions_mysql-1.1/src/set.c memcached_functions_mysql-1.1.patched/src/set.c --- memcached_functions_mysql-1.1/src/set.c 2009-11-11 10:56:52.000000000 +0800 +++ memcached_functions_mysql-1.1.patched/src/set.c 2015-09-19 10:01:36.611512903 +0800 @@ -30,7 +30,7 @@ unsigned int count; memc_function_st *container; - container= prepare_args(args, message, MEMC_SET, 2, 3); + container= prepare_args(args, message, MEMC_SET, 2, 4); if (container == NULL) return 1; @@ -67,7 +67,7 @@ rc= memcached_set(&container->memc, args->args[0], (size_t)args->lengths[0], args->args[1], (size_t)args->lengths[1], - container->expiration, (uint16_t)0); + container->expiration, container->flags); return (rc != MEMCACHED_SUCCESS) ? (long long) 0 : (long long) 1; }如此调整后memc_set方法的第三个参数是flags，第四个参数是过期时间，不传都默认为0。
由于PHP Memcached模块中会对保存的数据进行如序列化、压缩等处理，而在同步时UDF很难做到一样的处理方法，因此需要将PHP Memcached模块处理后的数据获取出来，我们在PHP Memcached模块中增加一个getPayload的方法来实现这一功能，其实现如下
static zend_function_entry memcached_class_methods[] = { //... MEMC_ME(getPayload, arginfo_getPayload) { NULL, NULL, NULL } }; ZEND_BEGIN_ARG_INFO_EX(arginfo_getPayload, 0, 0, 1) ZEND_ARG_INFO(0, value) ZEND_ARG_INFO(1, flags) ZEND_END_ARG_INFO() static PHP_METHOD(Memcached, getPayload) { zval* zflags = NULL; uint32_t flags; zval* value; char *payload; size_t payload_len; MEMC_METHOD_INIT_VARS; if (zend_parse_parameters(ZEND_NUM_ARGS() TSRMLS_CC, "z|z", &value, &zflags) == FAILURE) { return; } MEMC_METHOD_FETCH_OBJECT; if (i_obj->compression) { flags |= MEMC_VAL_COMPRESSED; } if (i_obj->serializer == SERIALIZER_IGBINARY) { flags |= MEMC_VAL_IGBINARY; } payload = php_memc_zval_to_payload(value, &payload_len, &flags TSRMLS_CC); if (zflags != NULL) { convert_to_long(zflags); ZVAL_LONG(zflags, flags); } RETURN_STRINGL(payload, payload_len, 0); }其使用效果如下
<?php $mem = new Memcached(); var_dump($mem->getPayload(array(), $flags), $flags);输入如下
string(6) "a:0:{}" int(1)
建立用于同步的表，其表结构如下
CREATE TABLE `MemcQueue` ( `k` varchar(250) NOT NULL, `expiration` int(11) NOT NULL, `flags` int(10) unsigned NOT NULL, `v` mediumblob NOT NULL, PRIMARY KEY (`k`) ) ENGINE=BLACKHOLE DEFAULT CHARSET=latin1在从库中建立触发器
DELIMITER | CREATE TRIGGER queue_trigger AFTER INSERT ON MemcQueue FOR EACH ROW BEGIN SET @mm = memc_set(new.k, new.v, new.flags, new.expiration); END |
测试使用脚本如下
<?php $key = 'test'; $expire = 600; $pdo = new PDO('mysql:host=127.0.0.1;dbname=test', 'root', ''); $stat = $pdo->prepare("INSERT INTO MemcQueue VALUES (?, ?, ?, ?)"); $mem = new Memcached(); $mem->setOption(Memcached::DISTRIBUTION_CONSISTENT, 1); $mem->setOption(Memcached::OPT_LIBKETAMA_COMPATIBLE, 1); $mem->addServers(array(array('127.0.0.1', 11211), array('127.0.0.1', 11212))); $v = array(); for ($i = 0; $i < 5; $i++) { $v[] = $i; } $payload = $mem->getPayload($v, $flags); for ($i = 0; $i < 10000; $i++) { $stat->execute(array($key, $expire, $flags, $payload)); } var_dump($mem->get($key)); ?>执行效果如下
time php mem.php array(5) { [0]=> int(0) [1]=> int(1) [2]=> int(2) [3]=> int(3) [4]=> int(4) } real 0m2.141s user 0m0.105s sys 0m0.131s
数独(Sudoku)是一种非常简单的数字游戏， 对于经典的数独其规则是根据给出的数字在一个9X9的方格中填写剩余方格中的数字，需要保证每一行的数字包含1~9，每一列包含1~9，每个3X3的子方格中也包含1~9。关于数独的各种理论分析这里不做过多的讨论，而只是给出一种简单并且适合计算机求解的方法。
首先对于一个给出的题目，其可能存在唯一解，可能存在多个解，也可能无解，我们的算法则是在有解的情况下给出其中的一个解答。
首先根据已有的数字，利用上面提到的三个约束条件来确定可以唯一确定数字的方格，并将其填入相应的数据
对于其他无法确定的方格，选取其中一个，并利用上面提到的三个约束条件找出可能填写的数字，并尝试其中的一个，如此就形成一个新的数独问题，对此数独问题按照求解流程进行递归求解
如果递归求解成功，则数独问题求解完成，如果求解失败（如果存在一个方格中没有能够填写的数字或是已经将一个方格中的全部可能的数字都尝试后依然无法完成求解），则继续尝试下一个可能的数字
如果全部数字都无法完成求解则此数独无解
求解的过程是一个典型的深度优先搜索问题，程序中利用栈来保存当前的状态并对每一个分支进行分析，如果分析失败则回溯到最近一次栈中的状态，实际过程中可以利用函数递归调用中的堆栈来进行状态的保存，当然也可以自己实现一个栈的数据结构实现非递归的算法。
根据以上的求解思路，这里给出一种Java语言的实现，其代码如下
import java.io.*; import java.util.Scanner; import java.util.ArrayList; class Sudoku { private int[][] matrix; final int SIZE = 9; public Sudoku() { matrix = new int[SIZE][SIZE]; } //复制一个数独，便于新的数独问题的生成 public Sudoku(Sudoku s) { this(); int size = SIZE; for (int i = 0; i < size; i++) { for (int j = 0; j < size; j++) { set(i, j, s.get(i, j)); } } } //读写数独方格 public void set(int i, int j, int v) { matrix[i][j] = v; } public int get(int i, int j) { return matrix[i][j]; } //根据目前的数字，利用约束填写其他的方格直到无法找到可以填写的方格为止 private boolean calculate() { int size = SIZE; int n; //遍历全部的方法，如果能够找到可以填写的方格，则进行进行遍历知道无法找到为止 do { n = 0; for (int i = 0; i < size; i++) { for (int j = 0; j < size; j++) { //如果已经填写了数字则不进行分析 if (get(i, j) != 0) { continue; } //获取可以填写的数字 ArrayList<Integer> nums = getAvailNums(i, j); int s = nums.size(); //如果是唯一的，则进行填写，并记录找到一个方格 if (s == 1) { int v = nums.get(0); set(i, j, v); n++; } //如果找到一个不能填写任何数字的方格则返回求解失败 if (s == 0) { return false; } } } } while (n > 0); //遍历完成返回遍历成功 return true; } //数独求解主流程 public boolean solve() { int size = SIZE; //根据约束条件填写方格，如果失败则整个求解流程失败 if (!calculate()) { return false; } //遍历整个方格没有填写的方格，直到全部填写完成为止 int n; do { n = 0; for (int i = 0; i < size; i++) { for (int j = 0; j < size; j++) { //如果已经填写则增加计数 if (get(i, j) != 0) { n++; continue; } //直到当前方格中可以填写的数字 ArrayList<Integer> nums = getAvailNums(i, j); int s = nums.size(); //如果已经没有可以填写的数字，则返回求解失败 if (s == 0) { return false; } //对于每一个可以填写的数字，生成一个新的数独问题，递归进行求解 for (int k = 0; k < s; k++) { Sudoku ns = new Sudoku(this); ns.set(i, j, nums.get(k)); //如果求解成功，则保存答案，返回成功 if (ns.solve()) { for (int p = 0; p < size; p++) { for (int q = 0; q < size; q++) { set(p, q, ns.get(p, q)); } } return true; } } //如果全部尝试后依然无法得到解答，则求解失败 return false; } } } while (n != size * size); //全部填满则返回求解成功 return true; } //获取当前方格中可以填写的数字 public ArrayList<Integer> getAvailNums(int i, int j) { int size = SIZE; //分析行和列 ArrayList<Integer> nums = getNums(); for (int k = 0; k < size; k++) { int index; int v1 = matrix[i][k]; if ((index = nums.indexOf(v1)) != -1) { nums.remove(index); } int v2 = matrix[k][j]; if ((index = nums.indexOf(v2)) != -1) { nums.remove(index); } } int ssize = (int) Math.sqrt(size); //分析子方格 int m = i / ssize; int n = j / ssize; for (int p = m * ssize; p < (m + 1) * ssize; p++) { for (int q = n * ssize; q < (n + 1) * ssize; q++) { int index; int v = matrix[p][q]; if ((index = nums.indexOf(v)) != -1) { nums.remove(index); } } } return nums; } private ArrayList<Integer> getNums() { int size = SIZE; ArrayList<Integer> nums = new ArrayList<Integer>(); for (int k = 0; k < size; k++) { nums.add(k + 1); } return nums; } //从文件加载数独 public void load(String file) { int size = SIZE; try { BufferedReader reader = new BufferedReader(new FileReader(new File(file))); for (int i = 0; i < size; i++) { String line = reader.readLine(); if (line == null) { break; } Scanner in = new Scanner(line); for (int j = 0; j < size; j++) { if (!in.hasNext()) { break; } matrix[i][j] = in.nextInt(); } } } catch (Exception ex) { ex.printStackTrace(); } } //输出数独方格情况 public void print() { int size = SIZE; for (int i = 0; i < SIZE; i++) { for (int j = 0; j < SIZE; j++) { System.out.printf("%d ", matrix[i][j]); } System.out.println(); } } //程序入口 public static void main(String[] args) { Sudoku s = new Sudoku(); s.load(args[0]); s.print(); System.out.println(); s.solve(); s.print(); } }
待求解数独，其中0代表需要填写的方格
0 0 0 0 0 0 0 0 2 0 0 4 2 0 0 6 0 1 6 0 0 0 0 0 9 0 0 9 6 0 8 0 4 1 0 0 0 0 0 9 0 3 0 0 0 0 0 8 7 0 6 0 4 9 0 0 5 0 0 0 0 0 8 1 0 7 0 0 8 3 0 0 4 0 0 0 0 0 0 0 0求解后的结果
8 1 9 4 6 5 7 3 2 5 7 4 2 3 9 6 8 1 6 2 3 1 8 7 9 5 4 9 6 2 8 5 4 1 7 3 7 4 1 9 2 3 8 6 5 3 5 8 7 1 6 2 4 9 2 3 5 6 7 1 4 9 8 1 9 7 5 4 8 3 2 6 4 8 6 3 9 2 5 1 7