vertx的官方自我介紹

Eclipse Vert.x is a tool-kit for building reactive applications on the JVM.

也就是說vertx是一個(gè)JVM上面的響應(yīng)式工具集肆饶。

之所以選擇vertx主要是考慮到springboot那一套過于笨重了纱意，另外在響應(yīng)式的支持上也不是很完善凯旭。

引入依賴

        <dependency>
            <groupId>io.vertx</groupId>
            <artifactId>vertx-kafka-client</artifactId>
            <version>3.8.3</version>
            <exclusions><!--去掉log4j的依賴，改為logback -->
                <exclusion>
                    <groupId>org.slf4j</groupId>
                    <artifactId>slf4j-log4j12</artifactId>
                </exclusion>
                <exclusion>
                    <groupId>log4j</groupId>
                    <artifactId>log4j</artifactId>
                </exclusion>
            </exclusions>
        </dependency>

                <!--lombok 一個(gè)簡化開發(fā)的利器-->
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <scope>provided</scope>
        </dependency>

        <!-- 反射工具包 -->
        <dependency>
            <groupId>org.reflections</groupId>
            <artifactId>reflections</artifactId>
            <version>0.9.11</version>
        </dependency>


        <!-- 導(dǎo)入slf4j的接口包以及對應(yīng)日志框架的驅(qū)動(dòng)包 -->
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-api</artifactId>
        </dependency>

        <dependency>
            <groupId>ch.qos.logback</groupId>
            <artifactId>logback-classic</artifactId>
        </dependency>

        <dependency>
            <groupId>ch.qos.logback</groupId>
            <artifactId>logback-core</artifactId>
        </dependency>
    </dependencies>

消費(fèi)者

import java.util.HashMap;
import java.util.Map;

import io.vertx.core.Vertx;
import io.vertx.kafka.client.consumer.KafkaConsumer;
import lombok.extern.slf4j.Slf4j;

@Slf4j
public class KafkaListener {

    public static final KafkaListener instance = new KafkaListener();

    KafkaConsumer<String, String> consumer;

    private KafkaListener() {
        Map<String, String> config = new HashMap<>();
        config.put("bootstrap.servers", "localhost:9092");
        config.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        config.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        config.put("group.id", "my_group");
        config.put("auto.offset.reset", "earliest");
        config.put("enable.auto.commit", "false");

        Vertx vertx = Vertx.vertx();
        // 創(chuàng)建一個(gè)Kafka Consumer
        consumer = KafkaConsumer.create(vertx, config);
    }

    public void listen(String topic) {
        consumer.subscribe(topic).handler(r -> {
            log.info("收到消息");
        });
    }
}

生產(chǎn)者

import java.util.Properties;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;

import io.vertx.core.Vertx;
import io.vertx.kafka.client.producer.KafkaProducer;
import io.vertx.kafka.client.producer.KafkaProducerRecord;
import io.vertx.kafka.client.producer.impl.KafkaProducerRecordImpl;
import lombok.extern.slf4j.Slf4j;

@Slf4j
public class KafkaSender {

    static KafkaProducer<String, String> producer;

    static {
        Vertx vertx = Vertx.vertx();
        Properties config = new Properties();
        config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        config.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        config.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        config.put(ConsumerConfig.GROUP_ID_CONFIG, "my_group");
        config.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        config.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");

        // 注意這里的第三和第四個(gè)參數(shù)
        producer = KafkaProducer.create(vertx, config, String.class, String.class);
    }

    @SuppressWarnings({ "rawtypes", "unchecked" })
    public static void send(String topic, String message) {
        KafkaProducerRecord<String, String> record = new KafkaProducerRecordImpl(topic, message);
        producer.send(record, f -> {
            log.info("kafka主題{}消息{}發(fā)送{}", topic, message, f.succeeded() ? "成功" : "失敗");
        });
    }
}
···