特定の MySQL テーブルの完全データと増分データをメッセージ キューに同期する - ソリューション

パッケージ org.kafka;

java.util.Arrays をインポートします。
java.util.Properties をインポートします。
org.apache.kafka.clients.consumer.ConsumerRecord をインポートします。
org.apache.kafka.clients.consumer.ConsumerRecords をインポートします。
org.apache.kafka.clients.consumer.KafkaConsumer をインポートします。
org.apache.kafka.common.serialization.StringDeserializer をインポートします。


/**
 *
 * タイトル: KafkaConsumerTest
 * 説明：
 * kafka コンシューマーデモ
 * バージョン:1.0.0
 * @著者 パンCM
 * @日付 2018年1月26日 */
パブリッククラス KafkaConsumerTest は Runnable を実装します {

    プライベート最終 KafkaConsumer<String, String> コンシューマー;
    プライベート ConsumerRecords<String, String> msgList;
    プライベート最終文字列トピック;
    プライベート静的最終文字列 GROUPID = "groupA";

    パブリック KafkaConsumerTest(文字列トピック名) {
        プロパティ props = new Properties();
        props.put("bootstrap.servers", "192.168.7.193:9092");
        props.put("group.id", GROUPID);
        props.put("enable.auto.commit", "true");
        props.put("auto.commit.interval.ms", "1000");
        props.put("セッションタイムアウト.ms", "30000");
        props.put("auto.offset.reset", "最新");
        props.put("key.deserializer", StringDeserializer.class.getName());
        props.put("value.deserializer", StringDeserializer.class.getName());
        this.consumer = 新しい KafkaConsumer<String, String>(props);
        this.topic = トピック名;
        this.consumer.subscribe(Arrays.asList(トピック));
    }

    @オーバーライド
    パブリックボイド実行() {
        int メッセージ番号 = 1;
        System.out.println("---------消費開始---------");
        試す {
            のために （; ; ） {
                メッセージリスト = consumer.poll(1000);
                if (null != msgList && msgList.count() > 0) {
                    (ConsumerRecord<String, String> レコード: msgList) {
                        // 100 件のレコードを消費した後に印刷しますが、印刷されたデータはこのパターンに従わない可能性があります System.out.println(messageNo + "=======receive: key = " + record.key() + ", value = " + record.value() + " offset===" + record.offset());


// 文字列 v = decodeUnicode(record.value());

// System.out.println(v);

                        // 1000 メッセージが消費されたら終了 if (messageNo % 1000 == 0) {
                            壊す;
                        }
                        メッセージNo++;
                    }
                } それ以外 {
                    スレッド.sleep(11);
                }
            }
        } キャッチ (InterruptedException e) {
            e.printStackTrace();
        ついに
            コンシューマーを閉じます。
        }
    }

    パブリック静的void main(String args[]) {
        KafkaConsumerTest test1 = 新しい KafkaConsumerTest("サンプルデータ");
        スレッド thread1 = new Thread(test1);
        スレッド1を開始します。
    }


    /*
     * 中国語をユニコードに変換*/
    パブリック静的文字列 gbEncoding(最終的な文字列 gbString) {
        char[] utfBytes = gbString.toCharArray();
        文字列 unicodeBytes = "";
        (int i = 0; i < utfBytes.length; i++) の場合 {
            文字列 hexB = Integer.toHexString(utfBytes[i]);
            （hexB.length() <= 2）の場合{
                hexB = "00" + hexB;
            }
            ユニコードバイト = ユニコードバイト + "\\u" + hexB;
        }
        unicodeBytes を返します。
    }

    /*
     * 中国語への Unicode エンコード */
    パブリック静的文字列decodeUnicode(final String dataStr) {
        開始 = 0;
        int 終了 = 0;
        最終的なStringBufferバッファ = 新しいStringBuffer();
        (開始 > -1) の間 {
            終了 = dataStr.indexOf("\\u", 開始 + 2);
            文字列 charStr = "";
            終了 == -1 の場合 {
                charStr = dataStr.substring(start + 2, dataStr.length());
            } それ以外 {
                charStr = dataStr.substring(開始 + 2、終了);
            }
            char letter = (char) Integer.parseInt(charStr, 16); // 16進整数文字列を解析します。
            buffer.append(新しいCharacter(letter).toString());
            開始 = 終了;
        }
        buffer.toString() を返します。

    }
}

テーブル `bak1` を作成します (
  `vin` varchar(20) NOT NULL,
  `p1` ダブルデフォルト NULL、
  `p2` ダブルデフォルト NULL、
  `p3` ダブルデフォルト NULL、
  `p4` ダブルデフォルト NULL、
  `p5` ダブルデフォルト NULL、
  `p6` ダブルデフォルト NULL、
  `p7` ダブルデフォルト NULL、
  `p8` ダブルデフォルト NULL、
  `p9` ダブルデフォルト NULL、
  `p0` ダブル デフォルト NULL
) エンジン=InnoDB デフォルト文字セット=utf8mb4

表示テーブル bak1 を作成します。

insert into bak1 select '李雷abcv',
  `p1`、
  `p2`、
  `p3`、
  `p4`、
  `p5`、
  `p6`、
  `p7`、
  `p8`、
  `p9`、
  moci 制限 10 からの `p0`