1

  Producer<String, String> producer = new KafkaProducer<String, String>(props);

1

2

3

4

5

6

7

8

9

10

11

12

13

====================================

類名：Sender

====================================

void run(long now) {

        if (transactionManager != null) {

            try {

                 ........

                if (!transactionManager.isTransactional()) {

                    // 為idempotent producer生成一個producer id

                    maybeWaitForProducerId();

                } else if (transactionManager.hasUnresolvedSequences() && !transactionManager.hasFatalError()) {

                   ........

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

private Producer buildIdempotProducer(){

        // create instance for properties to access producer configs

        Properties props = new Properties();

        // bootstrap.servers是Kafka叢集的IP地址。多個時,使用逗號隔開

        props.put("bootstrap.servers", "localhost:9092");

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

        //If the request fails, the producer can automatically retry,

        props.put("retries", 3);

        //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;

    }

1

2

3

4

5

6

7

public void produceIdempotMessage(String topic, String message) {

        // 建立Producer

        Producer producer = buildIdempotProducer();

        // 傳送訊息

        producer.send(new ProducerRecord<String, String>(topic, message));

        producer.flush();

    }

1

producer.initTransactions();

1

2

3

4

5

6

7

8

void  kakfa_in_tranction(){

  // 1.kafa的操作：讀取訊息或者生產訊息

kafkaOperation（）；

   // 2.db操作

  dbOperation()

}

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

/**

     * 初始化事務。需要注意的有：

     * 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 ;

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

/**

     * 在一個事務只有生產訊息操作

     */

    public void onlyProduceInTransaction() {

        Producer producer = buildProducer();

        // 1.初始化事務

        producer.initTransactions();

        // 2.開啟事務

        producer.beginTransaction();

        try {

            // 3.kafka寫操作集合

            // 3.1 do業務邏輯

            // 3.2 傳送訊息

            producer.send(new ProducerRecord<String, String>("test", "transaction-data-1"));

            producer.send(new ProducerRecord<String, String>("test", "transaction-data-2"));

            // 3.3 do其他業務邏輯,還可以傳送其他topic的訊息。

            // 4.事務提交

            producer.commitTransaction();

        } catch (Exception e) {

            // 5.放棄事務

            producer.abortTransaction();

        }

    }

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

36

37

38

39

40

41

42

43

44

45

46

47

/**

     * 需要:

     * 1、設定transactional.id

     * 2、設定enable.idempotence

     * @return

     */

    private Producer buildProducer() {

        // create instance for properties to access producer configs

        Properties props = new Properties();

        // bootstrap.servers是Kafka叢集的IP地址。多個時,使用逗號隔開

        props.put("bootstrap.servers", "localhost:9092");

        // 設定事務id

        props.put("transactional.id", "first-transactional");

        // 設定冪等性

        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;

    }

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

36

37

38

39

40

41

42

43

44

45

46

47

48

49

/**

     * 在一個事務內,即有生產訊息又有消費訊息

     */

    public void consumeTransferProduce() {

        // 1.構建上產者

        Producer producer = buildProducer();

        // 2.初始化事務(生成productId),對於一個生產者,只能執行一次初始化事務操作

        producer.initTransactions();

        // 3.構建消費者和訂閱主題

        Consumer consumer = buildConsumer();

        consumer.subscribe(Arrays.asList("test"));

        while (true) {

            // 4.開啟事務

            producer.beginTransaction();

            // 5.1 接受訊息

            ConsumerRecords<String, String> records = consumer.poll(500);

            try {

                // 5.2 do業務邏輯;

                System.out.println("customer Message---");

                Map<TopicPartition, OffsetAndMetadata> commits = Maps.newHashMap();

                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());

                    // 5.2.2 記錄提交的偏移量

                    commits.put(new TopicPartition(record.topic(), record.partition()),

                            new OffsetAndMetadata(record.offset()));

                    // 6.生產新的訊息。比如外賣訂單狀態的訊息,如果訂單成功,則需要傳送跟商家結轉訊息或者派送員的提成訊息

                    producer.send(new ProducerRecord<String, String>("test", "data2"));

                }

                // 7.提交偏移量

                producer.sendOffsetsToTransaction(commits, "group0323");

                // 8.事務提交

                producer.commitTransaction();

            } catch (Exception e) {

                // 7.放棄事務

                producer.abortTransaction();

            }

        }

    }

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

/**

     * 需要:

     * 1、關閉自動提交 enable.auto.commit

     * 2、isolation.level為

     * @return

     */

    public Consumer buildConsumer() {

        Properties props = new Properties();

        // bootstrap.servers是Kafka叢集的IP地址。多個時,使用逗號隔開

        props.put("bootstrap.servers", "localhost:9092");

        // 消費者群組

        props.put("group.id", "group0323");

        // 設定隔離級別

        props.put("isolation.level","read_committed");

        // 關閉自動提交

        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);

        return consumer;

    }

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

36

37

38

39

40

41

42

43

44

45

**

     * 在一個事務只有訊息操作

     */

    public void onlyConsumeInTransaction() {

        Producer producer = buildProducer();

        // 1.初始化事務

        producer.initTransactions();

        // 2.開啟事務

        producer.beginTransaction();

        // 3.kafka讀訊息的操作集合

        Consumer consumer = buildConsumer();

        while (true) {

            // 3.1 接受訊息

            ConsumerRecords<String, String> records = consumer.poll(500);

            try {

                // 3.2 do業務邏輯;

                System.out.println("customer Message---");

                Map<TopicPartition, OffsetAndMetadata> commits = Maps.newHashMap();

                for (ConsumerRecord<String, String> record : records) {

                    // 3.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());

                    // 3.2.2 記錄提交偏移量

                    commits.put(new TopicPartition(record.topic(), record.partition()),

                            new OffsetAndMetadata(record.offset()));

                }

                // 4.提交偏移量

                producer.sendOffsetsToTransaction(commits, "group0323");

                // 5.事務提交

                producer.commitTransaction();

            } catch (Exception e) {

                // 6.放棄事務

                producer.abortTransaction();

            }

        }

    }

Default is 604800000 (7 days). This allows periodic weekly producer jobs to maintain their ids.

Default is 900000 (15 min). This is a conservative upper bound on the period of time a transaction of messages will need to be sent.

Default: 3

Default: 50

Default: 2

Default: 104857600 bytes.

enable.idempotence

When idempotence is enabled, we enforce that acks=all, retries > 1, and max.inflight.requests.per.connection=1. Without these values for these configurations, we cannot guarantee idempotence. If these settings are not explicitly overidden by the application, the producer will set acks=all, retries=Integer.MAX_VALUE, and max.inflight.requests.per.connection=1 when idempotence is enabled.

This config value will be sent to the transaction coordinator along with the InitPidRequest. If this value is larger than the max.transaction.timeout.ms setting in the broker, the request will fail with a InvalidTransactionTimeout error.

Default is 60000. This makes a transaction to not block downstream consumption more than a minute, which is generally allowable in real-time apps.

如果配置了transactional.id屬性，則enable.idempotence 會被設定為true.

read_uncommitted: consume both committed and uncommitted messages in offset ordering.

read_committed: only consume non-transactional messages or committed transactional messages in offset order. In order to maintain offset ordering, this setting means that we will have to buffer messages in the consumer until we see all messages in a given transaction.

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

====================================

類名：ProducerFactoryUtils

====================================

/**

     * Obtain a Producer that is synchronized with the current transaction, if any.

     * @param producerFactory the ConnectionFactory to obtain a Channel for

     * @param <K> the key type.

     * @param <V> the value type.

     * @return the resource holder.

     */

    public static <K, V> KafkaResourceHolder<K, V> getTransactionalResourceHolder(

            final ProducerFactory<K, V> producerFactory) {

        Assert.notNull(producerFactory, "ProducerFactory must not be null");

        // 1.對於每一個執行緒會生成一個唯一key，然後根據key去查詢resourceHolder

        @SuppressWarnings("unchecked")

        KafkaResourceHolder<K, V> resourceHolder = (KafkaResourceHolder<K, V>) TransactionSynchronizationManager

                .getResource(producerFactory);

        if (resourceHolder == null) {

            // 2.建立一個消費者

            Producer<K, V> producer = producerFactory.createProducer();

            // 3.開啟事務

            producer.beginTransaction();

            resourceHolder = new KafkaResourceHolder<K, V>(producer);

            bindResourceToTransaction(resourceHolder, producerFactory);

        }

        return resourceHolder;

    }

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

====================================

類名：DefaultKafkaProducerFactory

====================================

protected Producer<K, V> createTransactionalProducer() {

Producer<K, V> producer = this.cache.poll();

if (producer == null) {

Map<String, Object> configs = new HashMap<>(this.configs);

            // 對於每一次生成producer時，都設定一個不同的transactionId

configs.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG,

this.transactionIdPrefix + this.transactionIdSuffix.getAndIncrement());

producer = new KafkaProducer<K, V>(configs, this.keySerializer, this.valueSerializer);

            // 1.初始化話事務。

producer.initTransactions();

return new CloseSafeProducer<K, V>(producer, this.cache);

}

else {

return producer;

}

}

1

2

3

4

5

6

AddPartitionsToTxnRequest => TransactionalId PID Epoch [Topic [Partition]]

TransactionalId => string

PID => int64

Epoch => int16

Topic => string

Partition => int32

1

2

3

4

5

AddOffsetsToTxnRequest => TransactionalId PID Epoch ConsumerGroupID

TransactionalId => string

PID => int64

Epoch => int16

ConsumerGroupID => string

1

2

3

4

5

6

7

8

9

10

11

12

13

14

TxnOffsetCommitRequest   => ConsumerGroupID

                            PID

                            Epoch

                            RetentionTime

                            OffsetAndMetadata

  ConsumerGroupID => string

  PID => int64

  Epoch => int32

  RetentionTime => int64

  OffsetAndMetadata => [TopicName [Partition Offset Metadata]]

    TopicName => string

    Partition => int32

    Offset => int64

    Metadata => string

1

2

3

4

5

6

7

WriteTxnMarkersRequest => [CoorinadorEpoch PID Epoch Marker [Topic [Partition]]]

CoordinatorEpoch => int32

PID => int64

Epoch => int16

Marker => boolean (false(0) means ABORT, true(1) means COMMIT)

Topic => string

Partition => int32

1

2

3

4

5

6

7

8

9

10

ReplicaId MaxWaitTime MinBytes [TopicName [Partition FetchOffset MaxBytes]]

  ReplicaId => int32

  MaxWaitTime => int32

  MinBytes => int32

  TopicName => string

  Partition => int32

  FetchOffset => int64

  MaxBytes => int32

  Isolation => READ_COMMITTED | READ_UNCOMMITTED

1

2

3

4

5

6

7

8

9

ThrottleTime => int32

  TopicName => string

  Partition => int32

  ErrorCode => int16

  HighwaterMarkOffset => int64

  AbortedTransactions => [PID FirstOffset]

    PID => int64

    FirstOffset => int64

  MessageSetSize => int32

1

2

3

4

5

Version => int16

PID => int64

FirstOffset => int64

LastOffset => int64

LastStableOffset => int64