@Service
public class ProductService {

    @Autowired
    private KafkaTemplate kafkaTemplate;
    
    public void send(String msg, String topic) {
        kafkaTemplate.send(topic, msg);
    }
}

/**
 * 初始化多分區(qū)的topic 基于springboot2
 */
@Component
public void TopicInitRunner implements ApplicationRunner {

    @Autowired
    private AdminClient adminClient;
    
    @Override
    public void run(ApplicationArguments args) throws Exception {
        // 通過配置文件讀取自定義配置的topic名及分區(qū)數(shù) 省略...
        // Key topic V 分區(qū)數(shù)
        Map<String, Integer> topicPartitionMap = new HashMap<>();
        for (Map.Entry<String, Integer> e : topicPartitionMap.entrySet()) {
            createTopic(e.getKey(), e.getValue());
        }
        
    }

    public void createTopic(String topic, int partition) {
        NewTopic newTopic = new NewTopic(topic, partition);
        adminClient.createTopics(Lists.newArrayList(newTopic));
    }
}

/**
 * 配置類參考 基于springboot2
 * 如果只進(jìn)行普通的單消息發(fā)送 無需添加此配置到項(xiàng)目中
 */
@Configuration
public class KafkaConfig {

    @Value("${spring.kafka.bootstrap-servers}")
    private String servers;

    @Bean
    public AdminClient adminClient() {
        return AdminClient.create(kafkaAdmin().getConfig());
    }

    @Bean
    public KafkaAdmin kafkaAdmin() {
        Map<String, Object> props = Maps.newHashMap();
        props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, servers);
        return new KafkaAdmin(props);
    }
}

public class ProductService {
    /**
     * data為需要發(fā)送的數(shù)據(jù)
     */
    public void partitionSend(String topic, int partition, JSONObject data) {
         // 獲取設(shè)備id
        String deviceId = data.getString("deviceId");
        // 獲取自增數(shù) 如果是新設(shè)備會(huì)創(chuàng)建一個(gè)并放入緩存中
        int inc = getDeviceInc(deviceId);
        // 如果分區(qū)數(shù)為3 設(shè)備自增id為1 取模結(jié)果為1 就是發(fā)送到1分區(qū) 這樣1000個(gè)設(shè)備就可以保證每個(gè)分區(qū)發(fā)送數(shù)據(jù)量是1000 / 3
        int targetPartition = Math.floorMod(inc, partition);
        // 分區(qū)發(fā)送時(shí)候 需要指定一個(gè)唯一k 可以使用uuid或者百度提供的雪花算法獲取id 字符串即可
        kafkaTemplate.send(topic, partition, getUuid(), data.toJSONString());
    }
}

@Component
public class DataListener {

    @Autowired
    private MongoTemplate mongoTemplate;

    /**
     * 站點(diǎn)報(bào)文監(jiān)聽消費(fèi)
     *
     * @param records
     */
    @KafkaListener(topics = "partition-topic", containerFactory = "batchConsumerFactory")
    public void iotSiteHistoryMessageConsumer(List<ConsumerRecord<String, String>> records) {        
    }
    
    /**
     * 消費(fèi)者配置
     */
    @Bean
    public Map<String, Object> consumerConfigs() {
        Map<String, Object> props = Maps.newHashMap();
        props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, maxPollRecords);
        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 120000);
        props.put(ConsumerConfig.REQUEST_TIMEOUT_MS_CONFIG, 180000);
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        return props;
    }
    
    /**
     * 批量消費(fèi)配置
     */
    @Bean
    public KafkaListenerContainerFactory batchConsumerFactory() {
        ConcurrentKafkaListenerContainerFactory<Integer, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(new DefaultKafkaConsumerFactory<>(consumerConfigs()));
        factory.setBatchListener(true);
        return factory;
    }
}