kafka-topics.sh --create --bootstrap-server server1:9092 --topic t_consumerGroup --partitions 2 --replication-factor 1

// 生产40条数据

import java.util.Properties;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;


public class KafkaProducerTest {
    // 只有主题分区超过一个以上的，才会由不同的消费者进行消费，一个分区的数据只能由组内的一个消费者进行消费，本示例的topic有2个分区
    public final static String TOPIC_NAME = "t_consumerGroup";

    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "server1:9092");
        props.put("acks", "all");
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        KafkaProducer<String, String> producer = new KafkaProducer<String, String>(props);
        boolean flag = true;
        int count = 0;
        while (flag) {
            count++;

            for (int i = 0; i < 10; ++i) {
                try {
                    // 获取返回值Future，该对象封装了返回值
                    Future<RecordMetadata> future = producer.send(new ProducerRecord<String, String>(TOPIC_NAME, i + "", i + ""));
                    // 调用一个Future.get()方法等待响应
                    future.get();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } catch (ExecutionException e) {
                    e.printStackTrace();
                }
            }

            if (count > 3)
                flag = false;
        }
        // 5. 关闭生产者
        producer.close();
    }

}

// 设置消费者组的属性
props.setProperty("group.id", TOPIC_NAME);

import java.time.Duration;
import java.util.Arrays;
import java.util.Properties;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

/**
 * @author alanchan
 *
 */
public class KafkaConsumerTest {
    public final static String TOPIC_NAME = "t_consumerGroup";

    public static void main(String[] args) throws Exception {
        Properties props = new Properties();
        props.setProperty("bootstrap.servers", "server1:9092,server2:9092,server3:9092");
        // 消费者组（可以使用消费者组将若干个消费者组织到一起），共同消费Kafka中topic的数据
        // 每一个消费者需要指定一个消费者组，如果消费者的组名是一样的，表示这几个消费者是一个组中的
        props.setProperty("group.id", TOPIC_NAME);
        // 自动提交offset
        props.setProperty("enable.auto.commit", "true");
        // 自动提交offset的时间间隔
        props.setProperty("auto.commit.interval.ms", "1000");
        // 拉取的key、value数据的
        props.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

        KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(props);
        kafkaConsumer.subscribe(Arrays.asList(TOPIC_NAME));
        while (true) {
            // Kafka的消费者一次拉取一批的数据
            ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(Duration.ofSeconds(5));
            // 5.将将记录（record）的offset、key、value都打印出来
            for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {
                // 主题
                String topic = consumerRecord.topic();
                // offset：这条消息处于Kafka分区中的哪个位置
                long offset = consumerRecord.offset();
                // key\value
                String key = consumerRecord.key();
                String value = consumerRecord.value();
                int partition = consumerRecord.partition();
                System.out.println("topic: " + topic + "  partition:" + partition + " offset:" + offset + " key:" + key + " value:" + value);
            }
            Thread.sleep(1000);
        }
    }

}

props.put("enable.idempotence",true);

//producer提供的事务方法
/**
* 初始化事务。需要注意的有：
* 1、前提
*   需要保证transation.id属性被配置。
* 2、这个方法执行逻辑是：
*   （1）Ensures any transactions initiated by previous instances of the producer with the same
 transactional.id are completed. If the previous instance had failed with a transaction in
*      progress, it will be aborted. If the last transaction had begun completion, but not yet finished, this method awaits its completion.
*    （2）Gets the internal producer id and epoch, used in all future transactional messages issued by the producer.

*/
    public void initTransactions();
 
/**
 * 开启事务
 */
    public void beginTransaction() throws ProducerFencedException ;
 
/**
 * 为消费者提供的在事务内提交偏移量的操作
 */
    public void sendOffsetsToTransaction(Map<TopicPartition, OffsetAndMetadata> offsets,String consumerGroupId) throws ProducerFencedException ;
 
/**
 * 提交事务
 */
    public void commitTransaction() throws ProducerFencedException;
 
/**
 * 放弃事务，类似回滚事务的操作
 */
    public void abortTransaction() throws ProducerFencedException ;

// 配置事务的id，开启了事务会默认开启幂等性
props.put("transactional.id", "first-transactional");

// 1. 消费者需要设置隔离级别
props.put("isolation.level","read_committed");
//  2. 关闭自动提交
 props.put("enable.auto.commit", "false");

# 创建名为ods_user和dwd_user的主题
kafka-topics.sh --create --bootstrap-server server1:9092 --topic ods_user --partitions 1 --replication-factor 1
kafka-topics.sh --create --bootstrap-server server1:9092 --topic dwd_user --partitions 1 --replication-factor 1
# 生产数据到 ods_user
kafka-console-producer.sh --broker-list server1:9092 --topic ods_user
# 从dwd_user消费数据
kafka-console-consumer.sh --bootstrap-server server1:9092 --topic dwd_user --from-beginning  --isolation-level  read_committed

// 一、创建一个消费者来消费ods_user中的数据
public Consumer buildConsumer() {
    Properties props = new Properties();
    // bootstrap.servers是Kafka集群的IP地址。多个时,使用逗号隔开
    props.put("bootstrap.servers", KAFKA_SERVERS);
    // 消费者群组
    props.put("group.id", TOPIC_ODS_USER);
    // 设置隔离级别
    // 1、关闭自动提交 enable.auto.commit
    // 2、isolation.level为read_committed
    props.put("isolation.level", "read_committed");
    // 关闭自动提交
    // 在代码里面也不能使用手动提交commitSync( )或者commitAsync( )
    props.put("enable.auto.commit", "false");
    props.put("session.timeout.ms", "30000");
    props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
    props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
    KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);

    consumer.subscribe(Arrays.asList(TOPIC_ODS_USER));
    return consumer;
}

// 二、编写createProducer方法，用来创建一个带有事务配置的生产者
    // 1、设置transactional.id
    // 2、设置enable.idempotence
    private Producer buildProducer() {
        // create instance for properties to access producer configs
        Properties props = new Properties();
        // bootstrap.servers是Kafka集群的IP地址。多个时,使用逗号隔开
        props.put("bootstrap.servers", KAFKA_SERVERS);
        // 设置事务id
        props.put("transactional.id", TOPIC_DWD_USER);
        // 设置幂等性
        // 生产者不需要再配置enable.idempotence，因为如果配置了transaction.id，则此时enable.idempotence会被设置为true
        props.put("enable.idempotence", true);
        // Set acknowledgements for producer requests.
        props.put("acks", "all");
        // If the request fails, the producer can automatically retry,
        props.put("retries", 1);
        // Specify buffer size in
        // config,这里不进行设置这个属性,如果设置了,还需要执行producer.flush()来把缓存中消息发送出去
        // props.put("batch.size", 16384);
        // Reduce the no of requests less than 0
        props.put("linger.ms", 1);
        // The buffer.memory controls the total amount of memory available to the
        // producer for buffering.
        props.put("buffer.memory", 33554432);
        // Kafka消息是以键值对的形式发送,需要设置key和value类型序列化器
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        Producer<String, String> producer = new KafkaProducer<String, String>(props);
        return producer;
    }

import java.time.Duration;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.Future;

import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.TopicPartition;

/**
 * 功能： 
 1、消费 ods_user 队列中的数据 
 2、在消费ods_user 队列中的数据同时，完成转换，同时将数据写入 dwd_user 队列中 
 验证：
 1、通过客户端向ods_user中写入业务数据 
 2、查看dwd_user中的数据是否按照要求进行转换
 * 
 * @author alanchan
 *
 */
public class ConsumeTransferProduceTransactionTest {
    private final static String TOPIC_ODS_USER = "ods_user";
    private final static String TOPIC_DWD_USER = "dwd_user";
    private final static String KAFKA_SERVERS = "server1:9092";

    // 一、创建一个消费者来消费ods_user中的数据
    public Consumer buildConsumer() {
        Properties props = new Properties();
        // bootstrap.servers是Kafka集群的IP地址。多个时,使用逗号隔开
        props.put("bootstrap.servers", KAFKA_SERVERS);
        // 消费者群组
        props.put("group.id", TOPIC_ODS_USER);
        // 设置隔离级别
        // 1、关闭自动提交 enable.auto.commit
        // 2、isolation.level为read_committed
        props.put("isolation.level", "read_committed");
        // 关闭自动提交
        // 在代码里面也不能使用手动提交commitSync( )或者commitAsync( )
        props.put("enable.auto.commit", "false");
        props.put("session.timeout.ms", "30000");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);

        consumer.subscribe(Arrays.asList(TOPIC_ODS_USER));
        return consumer;
    }

    // 二、编写createProducer方法，用来创建一个带有事务配置的生产者
    // 1、设置transactional.id
    // 2、设置enable.idempotence
    private Producer buildProducer() {
        // create instance for properties to access producer configs
        Properties props = new Properties();
        // bootstrap.servers是Kafka集群的IP地址。多个时,使用逗号隔开
        props.put("bootstrap.servers", KAFKA_SERVERS);
        // 设置事务id
        props.put("transactional.id", TOPIC_DWD_USER);
        // 设置幂等性
        // 生产者不需要再配置enable.idempotence，因为如果配置了transaction.id，则此时enable.idempotence会被设置为true
        props.put("enable.idempotence", true);
        // Set acknowledgements for producer requests.
        props.put("acks", "all");
        // If the request fails, the producer can automatically retry,
        props.put("retries", 1);
        // Specify buffer size in
        // config,这里不进行设置这个属性,如果设置了,还需要执行producer.flush()来把缓存中消息发送出去
        // props.put("batch.size", 16384);
        // Reduce the no of requests less than 0
        props.put("linger.ms", 1);
        // The buffer.memory controls the total amount of memory available to the
        // producer for buffering.
        props.put("buffer.memory", 33554432);
        // Kafka消息是以键值对的形式发送,需要设置key和value类型序列化器
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        Producer<String, String> producer = new KafkaProducer<String, String>(props);
        return producer;
    }

    // 三、将 TOPIC_ODS_USER 队列中user信息中的0/1转换成女/男，并写入 TOPIC_DWD_USER 队列
    //在一个事务内,即有生产消息又有消费消息，即常说的Consume-tansform-produce模式
    public void consumeTransferProduce() {
        // 1.构建生产者
        Producer producer = buildProducer();
        // 2.初始化事务(生成productId),对于一个生产者,只能执行一次初始化事务操作
        producer.initTransactions();
        // 3.构建消费者和订阅主题
        Consumer consumer = buildConsumer();

        while (true) {
            // 4.开启事务
            producer.beginTransaction();
            // 5.1 接受消息
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(5));
            try {
                // 5.2 do业务逻辑;
                Map<TopicPartition, OffsetAndMetadata> commits = new HashMap<>();

                for (ConsumerRecord<String, String> record : records) {
                    // 5.2.1 读取消息,并处理消息。print the offset,key and value for the consumer records.
                    System.out.printf("接收消息信息：offset = %d, key = %s, value = %s\n", record.offset(), record.key(), record.value());
                    String msg = trans(record.value());
                    System.out.println("转换后的消息信息：" + msg);

                    // 5.2.2 记录提交的偏移量
                    // offset + 1：offset是当前消费的记录（消息）对应在partition中的offset
                    // 必须要+1，解决服务重启后，消息消费不重复，即消费下一条消息
                    commits.put(new TopicPartition(record.topic(), record.partition()), new OffsetAndMetadata(record.offset() + 1));

                    // 6.生产新的消息。比如外卖订单状态的消息,如果订单成功,则需要发送跟商家结转消息或者派送员的提成消息
                    producer.send(new ProducerRecord<String, String>(TOPIC_DWD_USER, msg));
                    
                }

                // 7.提交偏移量
                producer.sendOffsetsToTransaction(commits, TOPIC_ODS_USER);
//                int i = 1 / 0;
                // 8.事务提交
                producer.commitTransaction();

            } catch (Exception e) {
                e.printStackTrace();
                // 7.放弃事务
                producer.abortTransaction();
            }
        }
    }

    public static void main(String[] args) {
        ConsumeTransferProduceTransactionTest ctp = new ConsumeTransferProduceTransactionTest();
        ctp.consumeTransferProduce();
    }

    private String trans(String message) {
        // 将1转换为男，0转换为女
        String[] fieldArray = message.split(",");

        // 将字段进行替换
        if (fieldArray != null && fieldArray.length > 2) {
            String sexField = fieldArray[1];
            if (sexField.equals("1")) {
                fieldArray[1] = "男";
            } else if (sexField.equals("0")) {
                fieldArray[1] = "女";
            }
        }

        // 重新拼接字段
        return fieldArray[0] + "," + fieldArray[1] + "," + fieldArray[2];
    }
}

[alanchan@server2 bin]$ kafka-console-producer.sh --broker-list server1:9092 --topic ods_user
>张三,1,1980-10-09
>

[alanchan@server3 bin]$ kafka-console-consumer.sh --bootstrap-server server1:9092 --topic dwd_user --from-beginning  --isolation-level  read_committed
张三,男,1980-10-09

kafka-leader-election.sh
--bootstrap-server server1:9092 
--topic 主题 
--partition=1
--election-type preferred

kafka-leader-election.sh
--bootstrap-server server1:9092 
--topic test 
--partition=2
--election-type preferred

# --partition：指定需要重新分配leader的partition编号

kafka-configs.sh
--bootstrap-server server1:9092
--alter 
--add-config 'producer_byte_rate=1048576' 
--entity-type clients 
--entity-default

kafka-producer-perf-test.sh 
--topic test
--num-records 500000 
--throughput -1 
--record-size 1000 
--producer-props
bootstrap.servers=server1:9092,server2:9092,server3:9092
acks=1

kafka-configs.sh
--bootstrap-server server1:9092 
--alter 
--add-config 'consumer_byte_rate=1048576' 
--entity-type clients 
--entity-default

kafka-consumer-perf-test.sh 
--broker-list server1:9092,server2:9092,server3:9092
--topic test
--fetch-size 1048576 
--messages 500000

kafka-configs.sh 
--bootstrap-server server1:9092  
--alter 
--delete-config 'producer_byte_rate' 
--entity-type  clients 
--entity-default

kafka-configs.sh 
--bootstrap-server server1:9092  
--alter 
--delete-config 'consumer_byte_rate' 
--entity-type  clients 
--entity-default