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

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

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应用包括以下这些部分:


 

QQ图片20220427164352.jpg


  • 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

相关文章
|
关系型数据库 MySQL 数据库
RDS用多了,你还知道MySQL主从复制底层原理和实现方案吗?
随着数据量增长和业务扩展,单个数据库难以满足需求,需调整为集群模式以实现负载均衡和读写分离。MySQL主从复制是常见的高可用架构,通过binlog日志同步数据,确保主从数据一致性。本文详细介绍MySQL主从复制原理及配置步骤,包括一主二从集群的搭建过程,帮助读者实现稳定可靠的数据库高可用架构。
1108 9
RDS用多了,你还知道MySQL主从复制底层原理和实现方案吗?
|
SQL 存储 关系型数据库
MySQL主从复制 —— 作用、原理、数据一致性,异步复制、半同步复制、组复制
MySQL主从复制 作用、原理—主库线程、I/O线程、SQL线程;主从同步要求,主从延迟原因及解决方案;数据一致性,异步复制、半同步复制、组复制
2162 11
|
监控 Java
java异步判断线程池所有任务是否执行完
通过上述步骤,您可以在Java中实现异步判断线程池所有任务是否执行完毕。这种方法使用了 `CompletionService`来监控任务的完成情况,并通过一个独立线程异步检查所有任务的执行状态。这种设计不仅简洁高效,还能确保在大量任务处理时程序的稳定性和可维护性。希望本文能为您的开发工作提供实用的指导和帮助。
611 17
|
JavaScript Java 中间件
Java CompletableFuture 异步超时实现探索
本文探讨了在JDK 8中`CompletableFuture`缺乏超时中断任务能力的问题,提出了一种异步超时实现方案,通过自定义工具类模拟JDK 9中`orTimeout`方法的功能,解决了任务超时无法精确控制的问题,适用于多线程并行执行优化场景。
630 0
|
SQL 关系型数据库 MySQL
Mysql中搭建主从复制原理和配置
主从复制在数据库管理中广泛应用,主要优点包括提高性能、实现高可用性、数据备份及灾难恢复。通过读写分离、从服务器接管、实时备份和地理分布等机制,有效增强系统的稳定性和数据安全性。主从复制涉及I/O线程和SQL线程,前者负责日志传输,后者负责日志应用,确保数据同步。配置过程中需开启二进制日志、设置唯一服务器ID,并创建复制用户,通过CHANGE MASTER TO命令配置从服务器连接主服务器,实现数据同步。实验部分展示了如何在两台CentOS 7服务器上配置MySQL 5.7主从复制,包括关闭防火墙、配置静态IP、设置域名解析、配置主从服务器、启动复制及验证同步效果。
652 0
Mysql中搭建主从复制原理和配置
|
存储 关系型数据库 MySQL
MySQL主从复制原理和使用
本文介绍了MySQL主从复制的基本概念、原理及其实现方法,详细讲解了一主两从的架构设计,以及三种常见的复制模式(全同步、异步、半同步)的特点与适用场景。此外,文章还提供了Spring Boot环境下配置主从复制的具体代码示例,包括数据源配置、上下文切换、路由实现及切面编程等内容,帮助读者理解如何在实际项目中实现数据库的读写分离。
1885 1
MySQL主从复制原理和使用
|
消息中间件 NoSQL 关系型数据库
一文彻底搞定Redis与MySQL的数据同步
【10月更文挑战第21天】本文介绍了 Redis 与 MySQL 数据同步的原因及实现方式。同步的主要目的是为了优化性能和保持数据一致性。实现方式包括基于数据库触发器、应用层双写和使用消息队列。每种方式都有其优缺点,需根据具体场景选择合适的方法。此外,文章还强调了数据同步时需要注意的数据一致性、性能优化和异常处理等问题。
3392 0
|
监控 关系型数据库 MySQL
深入了解MySQL主从复制:构建高效稳定的数据同步架构
深入了解MySQL主从复制:构建高效稳定的数据同步架构
581 1
JAVA并发编程系列(13)Future、FutureTask异步小王子
本文详细解析了Future及其相关类FutureTask的工作原理与应用场景。首先介绍了Future的基本概念和接口方法,强调其异步计算特性。接着通过FutureTask实现了一个模拟外卖订单处理的示例,展示了如何并发查询外卖信息并汇总结果。最后深入分析了FutureTask的源码,包括其内部状态转换机制及关键方法的实现原理。通过本文,读者可以全面理解Future在并发编程中的作用及其实现细节。
|
canal 消息中间件 关系型数据库
Canal作为一款高效、可靠的数据同步工具,凭借其基于MySQL binlog的增量同步机制,在数据同步领域展现了强大的应用价值
【9月更文挑战第1天】Canal作为一款高效、可靠的数据同步工具,凭借其基于MySQL binlog的增量同步机制,在数据同步领域展现了强大的应用价值
2557 4

推荐镜像

更多