请给出kafka的producer和consumer代码

好的，以下是Kafka的Producer和Consumer代码示例（使用Java语言）：

Producer代码示例：

import org.apache.kafka.clients.producer.*;

import java.util.Properties;

public class KafkaProducerExample {

    public static void main(String[] args) {

        String topicName = "test-topic"; // 指定topic名称

        Properties props = new Properties(); // 创建Properties对象
        props.put("bootstrap.servers", "localhost:9092"); // 指定Kafka Broker地址
        props.put("acks", "all"); // 指定Producer写入的确认方式
        props.put("retries", 0); // 如果发送失败，Producer重新发送的次数
        props.put("batch.size", 16384); // 批处理大小
        props.put("linger.ms", 1); // 等待时间，如果数据达到batch.size就不等待
        props.put("buffer.memory", 33554432); // 缓冲区大小

        // 创建Producer对象
        Producer<String, String> producer = new KafkaProducer<>(props);

        // 发送数据
        for (int i = 0; i < 10; i++)
            producer.send(new ProducerRecord<>(topicName, Integer.toString(i), Integer.toString(i)));

        System.out.println("Message sent successfully");

        producer.close();

    }
}

Consumer代码示例：

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.TopicPartition;
import java.util.Arrays;
import java.util.Properties;

public class KafkaConsumerExample {

    public static void main(String[] args) {

        String topicName = "test-topic"; // 指定topic名称
        String groupName = "test-group"; // 指定consumer group名称

        Properties props = new Properties(); // 创建Properties对象
        props.put("bootstrap.servers", "localhost:9092"); // 指定Kafka Broker地址
        props.put("group.id", groupName); // 指定consumer group
        props.put("enable.auto.commit", "true"); // 是否自动提交offset
        props.put("auto.commit.interval.ms", "1000"); // 自动提交offset的时间间隔
        props.put("session.timeout.ms", "30000"); // session超时设置
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); // key反序列化器
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); // value反序列化器

        // 创建Consumer对象
        KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);

        // 订阅topic
        consumer.subscribe(Arrays.asList(topicName));

        // 循环读取消息
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(100);
            for (ConsumerRecord<String, String> record : records) {
                System.out.println("offset = " + record.offset() + ", key = " + record.key() + ", value = " + record.value());

                // 手动提交offset
                TopicPartition tp = new TopicPartition(record.topic(), record.partition());
                OffsetAndMetadata oam = new OffsetAndMetadata(record.offset(), "no metadata");
                consumer.commitSync(Arrays.asList(new OffsetCommitRequest.TopicAndPartition[]{new OffsetCommitRequest.TopicAndPartition(tp, oam)}));
            }
        }
    }
}

希望这个代码示例对您有所帮助！