JAVA通过Gearman实现MySQL到Redis的数据同步(异步复制)

标签: java gearman mysql | 发表时间:2017-12-11 19:34 | 作者:huangjinjin520
出处:http://www.iteye.com
MySQL到Redis数据复制方案

无论MySQL还是Redis,自身都带有数据同步的机制,像比较常用的 MySQL的Master/Slave模式 ,就是由Slave端分析Master的binlog来实现的,这样的数据复制其实还是一个异步过程,只不过当服务器都在同一内网时,异步的延迟几乎可以忽略。

那么理论上我们也可以用同样方式,分析MySQL的binlog文件并将数据插入Redis。但是这需要对binlog文件以及MySQL有非常深入的理解,同时由于 binlog存在Statement/Row/Mixedlevel多种形式 ,分析binlog实现同步的工作量是非常大的。

因此这里选择了一种开发成本更加低廉的方式,借用已经比较成熟的MySQL UDF,将MySQL数据首先放入Gearman中,然后通过一个自己编写的PHP Gearman Worker,将数据同步到Redis。比分析binlog的方式增加了不少流程,但是实现成本更低,更容易操作。

Gearman的安装与使用

Gearman 是一个支持分布式的任务分发框架。设计简洁,获得了非常广泛的支持。一个典型的Gearman应用包括以下这些部分:






Gearman Job Server:Gearman核心程序,需要编译安装并以守护进程形式运行在后台
Gearman Client:可以理解为任务的收件员,比如我要在后台执行一个发送邮件的任务,可以在程序中调用一个Gearman Client并传入邮件的信息,然后就可以将执行结果立即展示给用户,而任务本身会慢慢在后台运行。
Gearman Worker:任务的真正执行者,一般需要自己编写具体逻辑并通过守护进程方式运行,Gearman Worker接收到Gearman Client传递的任务内容后,会按顺序处理。
以前曾经介绍过类似的 后台任务处理项目Resque 。两者的设计其实非常接近,简单可以类比为:

Gearman Job Server:对应Resque的Redis部分
Gearman Client:对应Resque的Queue操作
Gearman Worker:对应Resque的Worker和Job
这里之所以选择Gearman而不是Resque是因为Gearman提供了比较好用的MySQL UDF,工作量更小。


1、安装依赖
yum install -y boost-devel gperf libevent-devel libuuid-devel
yum install mysql-devel -y
2、下载gearman
wget https://launchpad.net/gearmand/1.2/1.1.12/+download/gearmand-1.1.12.tar.gz
3、编译安装,指定mysqlclient的链接路径
  tar -zxvf gearmand-1.1.12.tar.gz
  cd gearmand-1.1.12
   ./configure 
make && make install

4、启动gearmand服务端 (启动之时,在/var/log/下创建gearmand.log日志文件。-l 指定日志文件  -d后台运行 -L 0.0.0.0 绑定到IPV4
gearmand -L 0.0.0.0 -l /var/log/gearmand.log -d
5、查看是否启动成功
ps -ef | grep gearman
6、查看是否安装成功,查看gearman版本信息
gearmand -V

7、MySQL UDF + Trigger同步数据到Gearman (https://github.com/mysqludf)
安装lib_mysqludf_json(lib_mysqludf_json可以把MySQL表的数据以json数据格式输出)
wget https://github.com/mysqludf/lib_mysqludf_json/archive/master.zip
unzip master.zip
cd lib_mysqludf_json-master/
rm -rf lib_mysqludf_json.so
8、编译 mysql_config 这是mysql的配置文件,可以 find /usr -name mysql_config 搜索下在什么位置
gcc $(/usr/local/mysql/bin/mysql_config  --cflags) -shared -fPIC -o lib_mysqludf_json.so lib_mysqludf_json.c
9、拷贝lib_mysqludf_json.so到MySQL的plugin目录
(可以登陆MySQL,输入命令"show variables like '%plugin%'"查看plugin位置)
cp lib_mysqludf_json.so /usr/local/mysql/lib/plugin/

演示lib_mysqludf_json功能
登录mysql
mysql -uroot -h127.0.0.1 -p
注册UDF函数
CREATE FUNCTION json_object RETURNS STRING SONAME "lib_mysqludf_json.so";
CREATE FUNCTION json_array RETURNS STRING SONAME "lib_mysqludf_json.so";
CREATE FUNCTION json_members RETURNS STRING SONAME "lib_mysqludf_json.so";
CREATE FUNCTION json_values RETURNS STRING SONAME "lib_mysqludf_json.so";
//json_array|json_members|json_values函数注册方式与json_object一样.
select json_object(id,file_save_type,base_dir) as sys_file_save_config from sys_file_save_config;
ERROR 1123 (HY000): Can't initialize function 'json_object'; Invalid json member name - name cannot be empty
以上错误这样解决,给每个成员名称使用别名即可:
select json_object(id as id ,file_save_type as fileSaveType,app_id as appID) as sys_file_save_config from sys_file_save_config;


10、安装gearman-mysql-udf (https://launchpad.net/gearman-mysql-udf)
wget https://launchpad.net/gearman-mysql-udf/trunk/0.6/+download/gearman-mysql-udf-0.6.tar.gz
  tar zxvf gearman-mysql-udf-0.6.tar.gz
   cd gearman-mysql-udf-0.6
11、安装libgearman-devel
   yum install libgearman-devel -y
   如果没有yum源,添加epel.repo yum源
  [epel]
name=Extra Packages for Enterprise Linux 6 - $basearch
#baseurl=http://download.fedoraproject.org/pub/epel/6/$basearch
mirrorlist=https://mirrors.fedoraproject.org/metalink?repo=epel-6&arch=$basearch
failovermethod=priority
enabled=1
gpgcheck=1
gpgkey=file:///etc/pki/rpm-gpg/RPM-GPG-KEY-EPEL-6

[epel-debuginfo]
name=Extra Packages for Enterprise Linux 6 - $basearch - Debug
#baseurl=http://download.fedoraproject.org/pub/epel/6/$basearch/debug
mirrorlist=https://mirrors.fedoraproject.org/metalink?repo=epel-debug-6&arch=$basearch
failovermethod=priority
enabled=0
gpgkey=file:///etc/pki/rpm-gpg/RPM-GPG-KEY-EPEL-6
gpgcheck=1

[epel-source]
name=Extra Packages for Enterprise Linux 6 - $basearch - Source
#baseurl=http://download.fedoraproject.org/pub/epel/6/SRPMS
mirrorlist=https://mirrors.fedoraproject.org/metalink?repo=epel-source-6&arch=$basearch
failovermethod=priority
enabled=0
gpgkey=file:///etc/pki/rpm-gpg/RPM-GPG-KEY-EPEL-6
gpgcheck=1
   
12、编译安装
(可以登陆MySQL,输入命令"show variables like '%plugin%'"查看plugin位置, mysql_config的配置文件,以及插件库所在路径,编译之后会在此路径生成.so文件)
./configure --with-mysql=/usr/local/mysql/bin/mysql_config --libdir=/usr/local/mysql/lib/plugin/
make && make install

演示gearman-mysql-udf功能
mysql -uroot -p
CREATE FUNCTION gman_do_background RETURNS STRING SONAME "libgearman_mysql_udf.so";
CREATE FUNCTION gman_servers_set RETURNS STRING SONAME "libgearman_mysql_udf.so";
CREATE FUNCTION gman_do RETURNS STRING SONAME "libgearman_mysql_udf.so";
CREATE FUNCTION gman_do_high RETURNS STRING SONAME "libgearman_mysql_udf.so";
CREATE FUNCTION gman_do_low RETURNS STRING SONAME "libgearman_mysql_udf.so";
CREATE FUNCTION gman_do_high_background RETURNS STRING SONAME "libgearman_mysql_udf.so";
CREATE FUNCTION gman_do_low_background RETURNS STRING SONAME "libgearman_mysql_udf.so";
CREATE FUNCTION gman_sum RETURNS STRING SONAME "libgearman_mysql_udf.so";
//函数gman_do|gman_do_high|gman_do_low|gman_do_high_background|gman_do_low_background|gman_sum注册方式类似,请参考gearman-mysql-udf-0.6/README
//指定gearman job server地址
SELECT gman_servers_set('127.0.0.1:4730');

如果出现异常信息:
ERROR 1126 (HY000): Can't open shared library 'libgearman_mysql_udf.so' (errno: 11 libgearman.so.8: cannot open shared object file: No such file or directory)
表示系统找不到 libgearman.so 文件,一般so都在/usr/local/lib目录下,修改配置文件/etc/ld.so.conf,将/usr/local/lib目录加入进去即可:
$ cat /etc/ld.so.conf
include ld.so.conf.d/*.conf
/usr/local/lib
$ /sbin/ldconfig -v | grep gearman*

13、MySQL Trigger调用Gearman UDF实现同步
创建触发器
DELIMITER $$
CREATE TRIGGER test_data_to_redis AFTER UPDATE ON test FOR EACH ROW BEGIN
    SET@ret=gman_do_background('syncToRedis', json_object(NEW.id AS `id`, NEW.phone AS`phone`));
END$$;

DELIMITER $$
CREATE TRIGGER test_data_to_redis2 AFTER INSERT ON test
  FOR EACH ROW BEGIN
    SET @ret=gman_do_background('syncToRedis2', json_object(NEW.id AS `id`, NEW.phone AS`phone`));
  END$$
DELIMITER ;

DELIMITER $$
CREATE TRIGGER test_data_to_redis3 BEFORE DELETE ON test
  FOR EACH ROW BEGIN
    SET @ret=gman_do_background('syncToRedis3', json_object(OLD.id AS `id`, OLD.phone AS`phone`));
  END$$
DELIMITER ;

  说明以及问题:此类采用了gearman官网的java-gearman-service(地址:https://launchpad.net/gearman-java),目前release版本是0.6.6。java-gearman-servic.jar包中,即包括gearman server,还包括client和work客户端API。
  问题:config类为spring注入的配置文件类,在worker.addFunction中,如果通过config类的属性,并且属性是从配置文件来的就会有问题。不知道为啥,写死就是OK的。此类连接远程的gearman job server。
 
  jar包需要添加到本地jar仓库:
mvn install:install-file -Dfile=C:\software\java-gearman-service-0.6.6.jar -DgroupId=org.gearman.jgs -DartifactId=java-gearman-service -Dversion=0.6.6 -Dpackaging=jar

import java.util.concurrent.TimeUnit;

import org.gearman.Gearman;
import org.gearman.GearmanFunction;
import org.gearman.GearmanFunctionCallback;
import org.gearman.GearmanServer;
import org.gearman.GearmanWorker;

/**
* *ECHO_HOST = "192.168.125.131"为安装了Gearman并开启geramand服务的主机地址
*int ECHO_PORT = 4730默认端口为4730
*
* @author Administrator
*
*/
public class EchoWorker implements GearmanFunction {

// function name
public static final String ECHO_FUNCTION_NAME = "syncToRedis";

// job server地址
public static final String ECHO_HOST = "192.168.1.245";

// job server监听的端口
public static final int ECHO_PORT = 4730;

public static void main(String[] args) {
// 创建一个Gearman实例
Gearman gearman = Gearman.createGearman();
/*
* 创建一个jobserver
*
* Parameter 1: job server的IP地址 Parameter 2: job server监听的端口
*
* job server收到client的job,并将其分发给注册worker
*
*/
GearmanServer server = gearman.createGearmanServer(EchoWorker.ECHO_HOST, EchoWorker.ECHO_PORT);
// 创建一个Gearman的worker
GearmanWorker worker = gearman.createGearmanWorker(); // 正题来了,创建work节点。
worker.setReconnectPeriod(2, TimeUnit.SECONDS); // 设置超时重连时间
worker.setMaximumConcurrency(5); // 最大并发数
// 告诉工人如何执行工作(主要实现了GearmanFunction接口)
worker.addFunction(EchoWorker.ECHO_FUNCTION_NAME, new EchoWorker());
// worker连接服务器
worker.addServer(server);
}

@Override
public byte[] work(String function, byte[] data, GearmanFunctionCallback callback) throws Exception {
// work方法实现了GearmanFunction接口中的work方法,本实例中进行了字符串的反写
if (data != null) {
String str = new String(data);
System.out.println(str);
StringBuffer sb = new StringBuffer(str);
return sb.reverse().toString().getBytes();
} else {
return "未接收到data".getBytes();
}
}
}


import org.gearman.Gearman; 
import org.gearman.GearmanClient; 
import org.gearman.GearmanJobEvent; 
import org.gearman.GearmanJobReturn; 
import org.gearman.GearmanServer; 
 
public class EchoClient { 
    public static void main(String... args) throws InterruptedException { 
            //创建一个Gearman实例 
            Gearman gearman = Gearman.createGearman(); 
            //创建一个Gearman client              
            GearmanClient client = gearman.createGearmanClient(); 
            /* 
             * 创建一个jobserver 
             *  
             * Parameter 1: job server的IP地址 
             * Parameter 2: job server监听的端口 
             *  
             *job server收到client的job,并将其分发给注册worker 
             * 
             */ 
            GearmanServer server = gearman.createGearmanServer( 
                            EchoWorker.ECHO_HOST, EchoWorker.ECHO_PORT); 
             // 告诉客户端,提交工作时它可以连接到该服务器 
            client.addServer(server); 
            /* 
             * 向job server提交工作 
             *  
             * Parameter 1: gearman function名字 
             * Parameter 2: 传送给job server和worker的数据 
             *  
             * GearmanJobReturn返回job发热结果 
             */ 
            GearmanJobReturn jobReturn = client.submitJob( 
                            EchoWorker.ECHO_FUNCTION_NAME, ("Hello World!").getBytes()); 
            //遍历作业事件,直到我们打到最后文件              
            while (!jobReturn.isEOF()) { 
 
                    //下一个作业事件 
                    GearmanJobEvent event = jobReturn.poll(); 
 
                    switch (event.getEventType()) { 
 
                    case GEARMAN_JOB_SUCCESS:     //job执行成功 
                            System.out.println(new String(event.getData())); 
                            break; 
                    case GEARMAN_SUBMIT_FAIL:     //job提交失败 
                         
                    case GEARMAN_JOB_FAIL:        //job执行失败 
                            System.err.println(event.getEventType() + ": " 
                                            + new String(event.getData())); 
                    default: 
                    } 
            } 
            //关闭 
            gearman.shutdown(); 
    } 


http://gearman.org/download/
php方案:https://www.tuicool.com/articles/B7Jjaa








已有 0 人发表留言,猛击->> 这里<<-参与讨论


ITeye推荐



相关 [java gearman mysql] 推荐:

JAVA通过Gearman实现MySQL到Redis的数据同步(异步复制)

- - 企业架构 - ITeye博客
MySQL到Redis数据复制方案. 无论MySQL还是Redis,自身都带有数据同步的机制,像比较常用的 MySQL的Master/Slave模式 ,就是由Slave端分析Master的binlog来实现的,这样的数据复制其实还是一个异步过程,只不过当服务器都在同一内网时,异步的延迟几乎可以忽略.

mysql java hibernate类型对应

- - 企业架构 - ITeye博客
类型名称 显示长度 数据库类型 JAVA类型 JDBC类型索引(int) 描述   . CHAR N CHAR java.lang.String 1  字符型  . BLOB L+N BLOB java.lang.byte[] -4  二进制型  . TEXT 65535 VARCHAR java.lang.String -1 text文本型  .

Java通过mysql-connector-java-8.0.11连接MySQL Server 8.0遇到的几个问题

- - 编程语言 - ITeye博客
这次新安装了一个MySQL数据库,然后navicat连接数据库一点问题没有. 但是通过Java的jdbc连接却怎么都建立不了连接. 网上找了很久找到了原因:. 数据库用的是Mysql8版本,但工程里面mysql驱动包却是5.1.37版本. 只需修改驱动包为8.0.11版本即可. 而且驱动的包也改变了,由原来的:/generatorSqlmapCustom/lib/mysql-connector-java-5.1.28-bin.jar.

使用Java 8 Streams和Spring Data JPA流式传输MySQL结果

- -
2015年10月19日|  KrešimirNesek. 从1.8版开始,Spring数据项目包含一个有趣的功能 - 通过一个简单的API调用,开发人员可以请求将数据库查询结果作为Java 8流返回. 在技​​术上可行并且由底层数据库技术支持的情况下,结果将逐个流式传输,并且可以使用流操作进行处理.

基于glusterfs和gearman的离线任务运算分布式化方案介绍

- - 搜索研发部官方博客
web站点服务中,我们除了存在面向用户的服务功能外,往往也存在大量的后台离线的相关计算任务,如对前端的异步操作数据队列进行定期处理,对数据库中的数据进行汇总挖掘,监控,转储,对中间数据的进一步运算处理等等……一个web服务站点的背后,往往存在大量对应的后端处理任务的功能模块,用于支撑正常的业务功能系统.

正确使用MySQL JDBC setFetchSize()方法解决JDBC处理大结果集 java.lang.OutOfMemoryError: Java hea

- - Java - 编程语言 - ITeye博客
昨天在项目中需要对日志的查询结果进行导出功能. 日志导出功能的实现是这样的,输入查询条件,然后对查询结果进行导出. 之前的解决方案都是多次查询,然后使用limit 限制每次查询的条数. 那么能不能一次查询就把所有结果倒出来了. 于是我就使用一次查询,不使用limit分页. 结果出现 java.lang.OutOfMemoryError: Java heap space问题.

Linux Ksplice,MySQL and Oracle

- Syn - DBA Notes
Oracle 在 7 月份收购了 Ksplice. 使用了 Ksplice 的 Linux 系统,为 Kernel 打补丁无需重启动,做系统维护的朋友应该明白这是一个杀手级特性. 现在该产品已经合并到 Oracle Linux 中. 目前已经有超过 700 家客户,超过 10 万套系统使用了 Ksplice (不知道国内是否已经有用户了.

MySQL Replication 线程

- - CSDN博客推荐文章
Replication 线程. Mysql 的Replication 是一个异步的复制过程,从一个Mysql instace(我们称之为Master)复制到另一个Mysql instance(我们称之Slave). 在Master 与Slave 之间的实现整个复制过程主. 要由三个线程来完成,其中两个线程(Sql 线程和IO 线程)在Slave 端,另外一个线程(IO 线程)在Master 端.

mysql backup 脚本

- - ITeye博客
网上备份脚本很多,但考虑都不周全. 保证创建备份文件只能是创建者跟root可以访问,其他用户没有权限,保证了数据库备份的安全. 上面脚本是负责备份的份数管理,. 已有 0 人发表留言,猛击->> 这里<<-参与讨论. —软件人才免语言低担保 赴美带薪读研.

Oracle MySQL Or NoSQL续

- - Sky.Jian 朝阳的天空
接前面一篇,这里再将之前在“中国系统架构师大会”5周年的时候发布的纪念册“IT架构实录”上的一篇文章发出来,也算是前面博文中PPT的一个文字版解读吧. Oracle,MySQL 还是 NoSQL. 随着阿里系的“去IOE”运动在社区的宣传声越来越大,国内正在掀起一股“去xxx”的技术潮. 不仅仅是互联网企业,包括运营商以及金融机构都已经开始加入到这个潮流之中.