当您启动kafka代理时，您可以在中定义一组属性 conf/server.properties 文件。其中一个属性是 auto.create.topics.enable 如果您将此设置为true（默认情况下），当您向不存在的主题发送消息时，kafka将自动创建一个主题。分区号将由同一文件中的默认设置定义。
缺点：据我所知，以这种方式创建的主题将始终具有相同的默认设置（分区、副本…）。

更新时间：
kafka代理有一个属性：auto.create.topics.enable
如果将其设置为true，则生产者将消息发布到具有新主题名称的主题，它将自动为您创建主题。
confluent团队建议不要这样做，因为主题的爆炸（取决于您的环境）可能会变得很难处理，并且主题创建在创建时总是具有相同的默认值。复制系数至少为3是很重要的，以确保在磁盘出现故障时主题的持久性。

对于任何消息传递系统，我不认为建议使用producer动态创建主题/分区或任何队列的方法。
对于您的用例，您可能可以使用device\ id作为as分区键来区分消息，这样您就可以使用一个主题。

如果需要，可以从java创建一个主题。是否推荐，取决于用例。e、 g.如果您的主题名是生产者的传入有效负载的函数，那么它可能很有用。下面是在kafka0.10.x中工作的代码片段

void createTopic(String zookeeperConnect, String topicName) throws InterruptedException {
    int sessionTimeoutMs = <some-int-value>;
    int connectionTimeoutMs = <some-int-value>;

    ZkClient zkClient = new ZkClient(zookeeperConnect, sessionTimeoutMs, connectionTimeoutMs, ZKStringSerializer$.MODULE$);

    boolean isSecureKafkaCluster = false;
    ZkUtils zkUtils = new ZkUtils(zkClient, new  ZkConnection(zookeeperConnect), isSecureKafkaCluster);

    Properties topicConfig = new Properties();
    try {
      AdminUtils.createTopic(zkUtils, topicName, 1, 1, topicConfig,
      RackAwareMode.Disabled$.MODULE$);
    } catch (TopicExistsException ex) {
    //log it 
    }
    zkClient.close();
}

注意：只允许增加分区数。