@Test public void testConstructorWithSerializers() { Properties producerProps = new Properties(); producerProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9000"); new KafkaProducer<>(producerProps, new ByteArraySerializer(), new ByteArraySerializer()).close(); }
@Override public void close() { close(Long.MAX_VALUE, TimeUnit.MILLISECONDS); }
KafkaProducer implements Producer<K, V> { @Override public void close() { close(Long.MAX_VALUE, TimeUnit.MILLISECONDS); } }
KafkaProducer implements Producer<K, V> { @Override public void close() { close(Long.MAX_VALUE, TimeUnit.MILLISECONDS); } KafkaProducer(Map<String, Object> configs);  KafkaProducer(Map<String, Object> configs, Serializer<K> keySerializer, Serializer<V> valueSerializer);  KafkaProducer(Properties properties);  KafkaProducer(Properties properties, Serializer<K> keySerializer, Serializer<V> valueSerializer); @SuppressWarnings({"unchecked", "deprecation"}) private  KafkaProducer(ProducerConfig config, Serializer<K> keySerializer, Serializer<V> valueSerializer); }
KafkaProducer implements Producer<K, V> { @Override public void close() { close(Long.MAX_VALUE, TimeUnit.MILLISECONDS); } KafkaProducer(Map<String, Object> configs);  KafkaProducer(Map<String, Object> configs, Serializer<K> keySerializer, Serializer<V> valueSerializer);  KafkaProducer(Properties properties);  KafkaProducer(Properties properties, Serializer<K> keySerializer, Serializer<V> valueSerializer); @SuppressWarnings({"unchecked", "deprecation"}) private  KafkaProducer(ProducerConfig config, Serializer<K> keySerializer, Serializer<V> valueSerializer); void initTransactions(); void beginTransaction(); void sendOffsetsToTransaction(Map<TopicPartition, OffsetAndMetadata> offsets,
                                         String consumerGroupId); void commitTransaction(); void abortTransaction(); @Override Future<RecordMetadata> send(ProducerRecord<K, V> record); @Override // 发送消息，将消息放入RecordAccumulator暂存 Future<RecordMetadata> send(ProducerRecord<K, V> record, Callback callback); @Override void flush(); @Override // 从Metadata中获取指定的topc分区信息 List<PartitionInfo> partitionsFor(String topic); @Override Map<MetricName, ? extends Metric> metrics(); @Override void close(); @Override void close(long timeout, TimeUnit timeUnit); }
KafkaProducer implements Producer<K, V> { @Override public void close() { close(Long.MAX_VALUE, TimeUnit.MILLISECONDS); } KafkaProducer(Map<String, Object> configs);  KafkaProducer(Map<String, Object> configs, Serializer<K> keySerializer, Serializer<V> valueSerializer);  KafkaProducer(Properties properties);  KafkaProducer(Properties properties, Serializer<K> keySerializer, Serializer<V> valueSerializer); @SuppressWarnings({"unchecked", "deprecation"}) private  KafkaProducer(ProducerConfig config, Serializer<K> keySerializer, Serializer<V> valueSerializer); void initTransactions(); void beginTransaction(); void sendOffsetsToTransaction(Map<TopicPartition, OffsetAndMetadata> offsets,
                                         String consumerGroupId); void commitTransaction(); void abortTransaction(); @Override Future<RecordMetadata> send(ProducerRecord<K, V> record); @Override // 发送消息，将消息放入RecordAccumulator暂存 Future<RecordMetadata> send(ProducerRecord<K, V> record, Callback callback); @Override void flush(); @Override // 从Metadata中获取指定的topc分区信息 List<PartitionInfo> partitionsFor(String topic); @Override Map<MetricName, ? extends Metric> metrics(); @Override void close(); @Override void close(long timeout, TimeUnit timeUnit);  }