分布式运行模式下Kafka Producer的唯一事务ID

Question

我有一个基于过程消费的大数据应用程序-

假设我的应用程序在一台机器上运行，我实例化了2个消费者，他们有自己的生产者，例如生产者1有事务ID -

ExecutorService executorService// responsible for starting my consumers
for (int i = 0; i < 2; i++) {
    prod_trans_id = "consumer-" + str(i) + "-producer"
    Custom_Consumer consumer = new Custom_Consumer(prod_trans_id)
    executorService.submit(consumer)
}

如果我的应用程序在一台机器上工作，这完全可以正常工作，但是，事实并非如此，因为应用程序需要在多台机器上运行，因此当相同的代码在机器2上运行时，由机器2上的消费者实例化的生产者将具有与机器1上相同的事务ID。我希望事务ID的生成方式不会相互冲突，并且它们是可重现的，这意味着如果应用程序崩溃/停止（例如有人服务应用程序停止，然后服务应用程序启动），当它重新联机时，它应该使用以前使用的相同事务ID。我想到了基于UUID的方法，但是，UUID是随机的，当一台机器上的应用程序死亡并重新联机时，UUID将不相同。

Answer 1

private final static String HOSTNAME_COMMAND = "hostname";

public static String getHostName() {
        BufferedReader inputStreamReader = null;
        BufferedReader errorStreamReader = null;
        try {
            Process process = Runtime.getRuntime().exec(HOSTNAME_COMMAND);
            inputStreamReader = new BufferedReader(new InputStreamReader(process.getInputStream()));
            errorStreamReader = new BufferedReader(new InputStreamReader(process.getErrorStream()));
            if (errorStreamReader.readLine() != null) {
                throw new RuntimeException(String.format("Failed to get the hostname, exception message: %s",
                        errorStreamReader.readLine()));
            }
            return inputStreamReader.readLine();
        } catch (IOException e) {
            try {
                if (inputStreamReader != null) {
                    inputStreamReader.close();
                }
                if (errorStreamReader != null) {
                    errorStreamReader.close();
                }
            } catch (IOException e1) {
                LogExceptionTrace.logExceptionStackTrace(e1);
                throw new RuntimeException(e1);
            }
            LogExceptionTrace.logExceptionStackTrace(e);
            throw new RuntimeException(e);
        }
    }

然后使用主机名，如下所示:

final String producerTransactionalID = String.format("%s_producer", this.consumerName);

其中消费者名称设置如下：

for (int i = 0; i < NUMBER_OF_CONSUMERS; i++) {
            String consumerName = String.format("%s-worker-%d", hostName, i);
            Executor executor = new Executor(
                    Configuration, consumerName
            );
            Executors.add(executor);
            futures.add(executorService.submit(executor));
        }