kafka的版本问题

ゞ浴缸里的玫瑰 2023-02-17 05:12 11阅读 0赞

问题描述：项目中pom.xml关于kafka的依赖为如下所示：

<dependency>
        <groupId>org.apache.kafka</groupId>
        <artifactId>kafka_2.11</artifactId>
        <version>1.0.0</version>
    </dependency>

如果在项目中出现了kafka.javaapi.consumer.ConsumerConnector，则在引用ConsumerConnector类时会有如下提示：

kafka.javaapi.consumer.ConsumerConnector' is deprecated

解决办法：出现该问题的主要原因为ConsumerConnector类为kafka低版本中创建Consumer的接口，因此如果想使用低版本中的kafka可更换低版本的kafka依赖，低版本的依赖引入参考示例如下：

<dependency>
        <groupId>org.apache.kafka</groupId>
        <artifactId>kafka_2.9.2</artifactId>
        <version>0.8.1.1</version>
        <exclusions>
            <exclusion>
                <artifactId>jms</artifactId>
                <groupId>javax.jms</groupId>
            </exclusion>
            <exclusion>
                <artifactId>jmxtools</artifactId>
                <groupId>com.sun.jdmk</groupId>
            </exclusion>
            <exclusion>
                <artifactId>jmxri</artifactId>
                <groupId>com.sun.jmx</groupId>
            </exclusion>
        </exclusions>
    </dependency>

旧版本producer生成数据的参考示例：

import java.util.*;
    
    import kafka.javaapi.producer.Producer;
    import kafka.producer.KeyedMessage;
    import kafka.producer.ProducerConfig;
    
    public class SyncProduce {
    	public static void main(String[] args) { 
            Properties props = new Properties();
            props.put("metadata.broker.list", "ip1:port1,ip2:port2,ip3:port3");
            props.put("serializer.class", "kafka.serializer.StringEncoder");
    		//kafka.serializer.DefaultEncoder
            props.put("partitioner.class", "kafka.producer.partiton.SimplePartitioner");
    		//kafka.producer.DefaultPartitioner: based on the hash of the key
            props.put("request.required.acks", "1");
    		//0;  绝不等确认  1:   leader的一个副本收到这条消息，并发回确认 -1：   leader的所有副本都收到这条消息，并发回确认
     
            ProducerConfig config = new ProducerConfig(props);
     
            Producer<String, String> producer = new Producer<String, String>(config);
            Random rnd = new Random();
            for (int i = 0; i < 3; i++) { 
                   long runtime = new Date().getTime();  
                   String key = "key" + rnd.nextInt(255); 
                   String msg = runtime + "___" + key;
                   KeyedMessage<String, String> data = new KeyedMessage<String, String>("jiketest", key, msg);
    			   												//			 eventTopic, eventKey, eventBody
                   producer.send(data);
    			   try {
                       Thread.sleep(1000);
                   } catch (InterruptedException ie) {
    
                   }
            }
            producer.close();
        }
    }