Die vollständigen und inkrementellen Daten einer bestimmten MySQL-Tabelle mit der Nachrichtenwarteschlange synchronisieren - Lösung

Paket org.kafka;

importiere java.util.Arrays;
importiere java.util.Properties;
importiere org.apache.kafka.clients.consumer.ConsumerRecord;
importiere org.apache.kafka.clients.consumer.ConsumerRecords;
importiere org.apache.kafka.clients.consumer.KafkaConsumer;
importiere org.apache.kafka.common.serialization.StringDeserializer;


/**
 *
 * Titel: KafkaConsumerTest
 * Beschreibung:
 * Kafka-Consumer-Demo
 * Version: 1.0.0
 * @Autor pancm
 * @date 26. Januar 2018 */
öffentliche Klasse KafkaConsumerTest implementiert Runnable {

    privater finaler KafkaConsumer<String, String>-Verbraucher;
    private ConsumerRecords<String, String> msgList;
    privates endgültiges String-Thema;
    private statische endgültige Zeichenfolge GROUPID = "GruppeA";

    öffentlicher KafkaConsumerTest(String topicName) {
        Eigenschaften-Eigenschaften = neue Eigenschaften();
        props.put("bootstrap.servers", "192.168.7.193:9092");
        props.put("group.id", GRUPPENRID);
        props.put("aktivieren.auto.commit", "true");
        props.put("auto.commit.interval.ms", "1000");
        props.put("Sitzung.Timeout.ms", "30000");
        props.put("auto.offset.reset", "latest");
        props.put("key.deserializer", StringDeserializer.class.getName());
        props.put("Wert.deserializer", StringDeserializer.class.getName());
        this.consumer = neuer KafkaConsumer<String, String>(Eigenschaften);
        dieses.Thema = Themenname;
        this.consumer.subscribe(Arrays.asList(Thema));
    }

    @Überschreiben
    öffentliche Leere ausführen() {
        int Nachrichtennummer = 1;
        System.out.println("---------Verbrauch starten---------");
        versuchen {
            für (; ; ) {
                msgList = Verbraucher.Umfrage(1000);
                if (null != msgList && msgList.count() > 0) {
                    für (ConsumerRecord<String, String> Datensatz : msgList) {
                        //Drucken, nachdem 100 Datensätze verbraucht wurden, aber die gedruckten Daten folgen möglicherweise nicht diesem Muster System.out.println(messageNo + "=======receive: key = " + record.key() + ", value = " + record.value() + " offset===" + record.offset());


// Zeichenfolge v = decodeUnicode(Datensatz.Wert());

// System.out.println(v);

                        // Beenden, wenn 1000 Nachrichten verbraucht sind, if (messageNo % 1000 == 0) {
                            brechen;
                        }
                        NachrichtNr++;
                    }
                } anders {
                    Thread.sleep(11);
                }
            }
        } Fang (UnterbrocheneAusnahme e) {
            e.printStackTrace();
        Endlich
            Verbraucher.schließen();
        }
    }

    öffentliche statische void main(String args[]) {
        KafkaConsumerTest test1 = neuer KafkaConsumerTest("Beispieldaten");
        Thread thread1 = neuer Thread(test1);
        thread1.start();
    }


    /*
     * Chinesisch in Unicode konvertieren*/
    öffentliche statische Zeichenfolge gbEncoding(finale Zeichenfolge gbString) {
        char[] utfBytes = gbString.toCharArray();
        Zeichenfolge unicodeBytes = "";
        für (int i = 0; i < utfBytes.Länge; i++) {
            String hexB = Integer.toHexString(utfBytes[i]);
            wenn (hexB.Länge() <= 2) {
                hexB = "00" + hexB;
            }
            unicodeBytes = unicodeBytes + "\\u" + hexB;
        }
        UnicodeBytes zurückgeben;
    }

    /*
     * Unicode-Kodierung ins Chinesische*/
    öffentlicher statischer String decodeUnicode(finaler String dataStr) {
        Geben Sie den Startwert ein.
        int Ende = 0;
        endgültiger StringBuffer-Puffer = neuer StringBuffer();
        während (Start > -1) {
            Ende = dataStr.indexOf("\\u", Start + 2);
            Zeichenfolge charStr = "";
            wenn (Ende == -1) {
                charStr = dataStr.substring(start + 2, dataStr.length());
            } anders {
                charStr = dataStr.substring(start + 2, end);
            }
            char letter = (char) Integer.parseInt(charStr, 16); // Hexadezimale Ganzzahlzeichenfolge analysieren.
            Puffer.anhängen(neues Zeichen(Buchstabe).toString());
            Anfang = Ende;
        }
        gibt buffer.toString() zurück;

    }
}

TABELLE ERSTELLEN `bak1` (
  `vin` varchar(20) NICHT NULL,
  `p1` doppelter STANDARD NULL,
  `p2` doppelter STANDARD NULL,
  `p3` doppelter STANDARD NULL,
  `p4` doppelter DEFAULT NULL,
  `p5` doppelter DEFAULT NULL,
  `p6` doppelter DEFAULT NULL,
  `p7` doppelter DEFAULT NULL,
  `p8` doppelter DEFAULT NULL,
  `p9` doppelter DEFAULT NULL,
  `p0` doppelter STANDARD NULL
) ENGINE=InnoDB STANDARD-CHARSET=utf8mb4

zeige, erstelle Tabelle bak1;

in bak1 einfügen, wähle '李雷abcv',
  `p1`,
  `p2`,
  `p3`,
  `p4`,
  `p5`,
  `p6`,
  `p7`,
  `p8`,
  `p9`,
  `p0` von Moci-Grenze 10