JAVA通過(guò)Gearman實(shí)現(xiàn)MySQL到Redis的數(shù)據(jù)同步（異步復(fù)制）

doodlewind 發(fā)布于2019-08-16 10:56 / 1952人閱讀

摘要：但是這需要對(duì)文件以及有非常深入的理解，同時(shí)由于存在多種形式，分析實(shí)現(xiàn)同步的工作量是非常大的。因此這里選擇了一種開發(fā)成本更加低廉的方式，借用已經(jīng)比較成熟的，將數(shù)據(jù)首先放入中，然后通過(guò)一個(gè)自己編寫的，將數(shù)據(jù)同步到。此類連接遠(yuǎn)程的。

MySQL到Redis數(shù)據(jù)復(fù)制方案

無(wú)論MySQL還是Redis，自身都帶有數(shù)據(jù)同步的機(jī)制，像比較常用的 MySQL的Master/Slave模式，就是由Slave端分析Master的binlog來(lái)實(shí)現(xiàn)的，這樣的數(shù)據(jù)復(fù)制其實(shí)還是一個(gè)異步過(guò)程，只不過(guò)當(dāng)服務(wù)器都在同一內(nèi)網(wǎng)時(shí)，異步的延遲幾乎可以忽略。

那么理論上我們也可以用同樣方式，分析MySQL的binlog文件并將數(shù)據(jù)插入Redis。但是這需要對(duì)binlog文件以及MySQL有非常深入的理解，同時(shí)由于 binlog存在Statement/Row/Mixedlevel多種形式，分析binlog實(shí)現(xiàn)同步的工作量是非常大的。

因此這里選擇了一種開發(fā)成本更加低廉的方式，借用已經(jīng)比較成熟的MySQL UDF，將MySQL數(shù)據(jù)首先放入Gearman中，然后通過(guò)一個(gè)自己編寫的PHP Gearman Worker，將數(shù)據(jù)同步到Redis。比分析binlog的方式增加了不少流程，但是實(shí)現(xiàn)成本更低，更容易操作。

Gearman的安裝與使用

Gearman 是一個(gè)支持分布式的任務(wù)分發(fā)框架。設(shè)計(jì)簡(jiǎn)潔，獲得了非常廣泛的支持。一個(gè)典型的Gearman應(yīng)用包括以下這些部分：

Gearman Job Server：Gearman核心程序，需要編譯安裝并以守護(hù)進(jìn)程形式運(yùn)行在后臺(tái)
Gearman Client：可以理解為任務(wù)的收件員，比如我要在后臺(tái)執(zhí)行一個(gè)發(fā)送郵件的任務(wù)，可以在程序中調(diào)用一個(gè)Gearman Client并傳入郵件的信息，然后就可以將執(zhí)行結(jié)果立即展示給用戶，而任務(wù)本身會(huì)慢慢在后臺(tái)運(yùn)行。
Gearman Worker：任務(wù)的真正執(zhí)行者，一般需要自己編寫具體邏輯并通過(guò)守護(hù)進(jìn)程方式運(yùn)行，Gearman Worker接收到Gearman Client傳遞的任務(wù)內(nèi)容后，會(huì)按順序處理。
以前曾經(jīng)介紹過(guò)類似的后臺(tái)任務(wù)處理項(xiàng)目Resque 。兩者的設(shè)計(jì)其實(shí)非常接近，簡(jiǎn)單可以類比為：

Gearman Job Server：對(duì)應(yīng)Resque的Redis部分
Gearman Client：對(duì)應(yīng)Resque的Queue操作
Gearman Worker：對(duì)應(yīng)Resque的Worker和Job
這里之所以選擇Gearman而不是Resque是因?yàn)镚earman提供了比較好用的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、啟動(dòng)gearmand服務(wù)端（啟動(dòng)之時(shí)，在/var/log/下創(chuàng)建gearmand.log日志文件。-l 指定日志文件 -d后臺(tái)運(yùn)行 -L 0.0.0.0 綁定到IPV4

gearmand -L 0.0.0.0 -l /var/log/gearmand.log -d

5、查看是否啟動(dòng)成功

ps -ef | grep gearman

6、查看是否安裝成功，查看gearman版本信息

gearmand -V

7、MySQL UDF + Trigger同步數(shù)據(jù)到Gearman (https://github.com/mysqludf)
安裝lib_mysqludf_json（lib_mysqludf_json可以把MySQL表的數(shù)據(jù)以json數(shù)據(jù)格式輸出）

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

注冊(cè)UDF函數(shù)

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函數(shù)注冊(cè)方式與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
以上錯(cuò)誤這樣解決，給每個(gè)成員名稱使用別名即可：
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...

 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的配置文件，以及插件庫(kù)所在路徑，編譯之后會(huì)在此路徑生成.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"; 
//函數(shù)gman_do|gman_do_high|gman_do_low|gman_do_high_background|gman_do_low_background|gman_sum注冊(cè)方式類似，請(qǐng)參考gearman-mysql-udf-0.6/README 
//指定gearman job server地址 
SELECT gman_servers_set("127.0.0.1:4730");

如果出現(xiàn)異常信息：
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)
表示系統(tǒng)找不到 libgearman.so 文件，一般so都在/usr/local/lib目錄下，修改配置文件/etc/ld.so.conf，將/usr/local/lib目錄加入進(jìn)去即可:

$ cat /etc/ld.so.conf
    include ld.so.conf.d/*.conf
    /usr/local/lib
$ /sbin/ldconfig -v | grep gearman*

13、MySQL Trigger調(diào)用Gearman UDF實(shí)現(xiàn)同步
創(chuàng)建觸發(fā)器

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 ;

說(shuō)明以及問(wèn)題：此類采用了gearman官網(wǎng)的java-gearman-service（地址：https://launchpad.net/gearman...），目前release版本是0.6.6。java-gearman-servic.jar包中，即包括gearman server，還包括client和work客戶端API。
問(wèn)題：config類為spring注入的配置文件類，在worker.addFunction中，如果通過(guò)config類的屬性，并且屬性是從配置文件來(lái)的就會(huì)有問(wèn)題。不知道為啥，寫死就是OK的。此類連接遠(yuǎn)程的gearman job server。

jar包需要添加到本地jar倉(cāng)庫(kù)：

mvn install:install-file -Dfile=C:softwarejava-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服務(wù)的主機(jī)地址
 *int ECHO_PORT = 4730默認(rèn)端口為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監(jiān)聽的端口
public static final int ECHO_PORT = 4730;

public static void main(String[] args) {
    // 創(chuàng)建一個(gè)Gearman實(shí)例
    Gearman gearman = Gearman.createGearman();
    /*
     * 創(chuàng)建一個(gè)jobserver
     * 
     * Parameter 1: job server的IP地址 Parameter 2: job server監(jiān)聽的端口
     * 
     * job server收到client的job，并將其分發(fā)給注冊(cè)worker
     * 
     */
    GearmanServer server = gearman.createGearmanServer(EchoWorker.ECHO_HOST, EchoWorker.ECHO_PORT);
    // 創(chuàng)建一個(gè)Gearman的worker
    GearmanWorker worker = gearman.createGearmanWorker(); // 正題來(lái)了，創(chuàng)建work節(jié)點(diǎn)。
    worker.setReconnectPeriod(2, TimeUnit.SECONDS); // 設(shè)置超時(shí)重連時(shí)間
    worker.setMaximumConcurrency(5); // 最大并發(fā)數(shù)
    // 告訴工人如何執(zhí)行工作(主要實(shí)現(xiàn)了GearmanFunction接口)
    worker.addFunction(EchoWorker.ECHO_FUNCTION_NAME, new EchoWorker());
    // worker連接服務(wù)器
    worker.addServer(server);
}

@Override
public byte[] work(String function, byte[] data, GearmanFunctionCallback callback) throws Exception {
    // work方法實(shí)現(xiàn)了GearmanFunction接口中的work方法,本實(shí)例中進(jìn)行了字符串的反寫
    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 {  
        //創(chuàng)建一個(gè)Gearman實(shí)例  
        Gearman gearman = Gearman.createGearman();  
        //創(chuàng)建一個(gè)Gearman client               
        GearmanClient client = gearman.createGearmanClient();  
        /*  
         * 創(chuàng)建一個(gè)jobserver  
         *   
         * Parameter 1: job server的IP地址  
         * Parameter 2: job server監(jiān)聽的端口  
         *   
         *job server收到client的job，并將其分發(fā)給注冊(cè)worker  
         *  
         */  
        GearmanServer server = gearman.createGearmanServer(  
                        EchoWorker.ECHO_HOST, EchoWorker.ECHO_PORT);  
         // 告訴客戶端，提交工作時(shí)它可以連接到該服務(wù)器  
        client.addServer(server);  
        /*  
         * 向job server提交工作  
         *   
         * Parameter 1: gearman function名字  
         * Parameter 2: 傳送給job server和worker的數(shù)據(jù)  
         *   
         * GearmanJobReturn返回job發(fā)熱結(jié)果  
         */  
        GearmanJobReturn jobReturn = client.submitJob(  
                        EchoWorker.ECHO_FUNCTION_NAME, ("Hello World!").getBytes());  
        //遍歷作業(yè)事件，直到我們打到最后文件               
        while (!jobReturn.isEOF()) {  

                //下一個(gè)作業(yè)事件  
                GearmanJobEvent event = jobReturn.poll();  

                switch (event.getEventType()) {  

                case GEARMAN_JOB_SUCCESS:     //job執(zhí)行成功  
                        System.out.println(new String(event.getData()));  
                        break;  
                case GEARMAN_SUBMIT_FAIL:     //job提交失敗  
                      
                case GEARMAN_JOB_FAIL:        //job執(zhí)行失敗  
                        System.err.println(event.getEventType() + ": "  
                                        + new String(event.getData()));  
                default:  
                }  
        }  
        //關(guān)閉  
        gearman.shutdown();  
}

}
官網(wǎng)：http://gearman.org/download/
php方案：https://www.tuicool.com/artic...

跟多精彩文章