创建 Twitter 生产者

17 Mar 2025 | 4 分钟阅读

在本节中，我们将学习创建 Twitter 生产者。

创建 Twitter 生产者基本上有三个步骤

创建 Twitter 客户端。
创建生产者
发送推文

步骤 1： 创建一个新的 java 包，遵循包命名约定规则。然后，在其中创建一个 java 类，例如 'tweetproducer.java'。

步骤 2： 通过创建一个方法来创建 Twitter 客户端。现在，将 'github twitter java' 中的 Quickstart 代码复制到 Twitter 客户端方法中，如下所示

将其粘贴到新创建的方法中。此代码将在客户端和 hbc 主机之间创建连接。当队列为空或已满时，BlockingQueue 将阻止客户端对消息进行出队或入队。由于我们使用 hbc-core，因此只需要 msgQueue。另外，我们将关注术语，而不是人。因此，仅复制突出显示的代码。

现在，复制连接代码下方的 '创建客户端' 代码，如下所示

将代码粘贴到连接代码下方。此代码将通过客户端构建器创建一个 Twitter 客户端。由于我们使用的是 msgQueue，因此请勿复制红色突出显示的代码，该代码用于 eventMessageQueue。不需要它。

步骤 3： 以我们在前几节中学到的类似方式创建生产者，并建立引导服务器连接。

步骤 4： 创建 Kafka 生产者后，就可以将推文发送到 Kafka 了。从 'github twitter java' 复制 while 循环代码，该代码在 '创建客户端' 代码下方给出。粘贴到生产者代码下方。

现在，我们已准备好从 Twitter 读取推文。虽然，Kafka 生产者从主题读取消息。因此，请使用 CLI 上的 '-create' 命令创建指定的主题。另外，请指定分区值和复制因子。

例如：

这里，已创建主题 'twitter_topic'，其分区值为 6，复制因子为 1。最后，执行代码并在实际应用中体验 Kafka。

创建 Twitter 客户端的完整代码如下

package com.github.learnkafka;

import com.google.common.collect.Lists;
import com.twitter.hbc.ClientBuilder;
import com.twitter.hbc.core.Client;
import com.twitter.hbc.core.Constants;
import com.twitter.hbc.core.Hosts;
import com.twitter.hbc.core.HttpHosts;
import com.twitter.hbc.core.endpoint.StatusesFilterEndpoint;
import com.twitter.hbc.core.processor.StringDelimitedProcessor;
import com.twitter.hbc.httpclient.auth.Authentication;
import com.twitter.hbc.httpclient.auth.OAuth1;
import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;
import java.util.Properties;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

public class tweetproducer {
    Logger logger = LoggerFactory.getLogger(tweetproducer.class.getName());
    String consumerKey = "";//specify the consumer key from the twitter app
    String consumerSecret = "";//specify the consumerSecret key from the twitter app
    String token = "";//specify the token key from the twitter app
    String secret = "";//specify the secret key from the twitter app

    public tweetproducer() {}//constructor to invoke the producer function

    public static void main(String[] args) {
        new tweetproducer().run();
    }

    public void run() {
        logger.info("Setup");

        BlockingQueue<String> msgQueue = new         
	LinkedBlockingQueue<String>(1000);//Specify the size accordingly.
        Client client = tweetclient(msgQueue);
        client.connect(); //invokes the connection function
        KafkaProducer<String,String> producer=createKafkaProducer();

        // on a different thread, or multiple different threads....
        while (!client.isDone()) {
            String msg = null;
            try {
                msg = msgQueue.poll(5, TimeUnit.SECONDS);//specify the time
            } catch (InterruptedException e) {
                e.printStackTrace();
                client.stop();
            }
            if (msg != null) {
                logger.info(msg);
                producer.send(new ProducerRecord<>("twitter_topic", null, msg), new Callback() {
                    @Override
                    public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                        if(e!=null){
                            logger.error("Something went wrong",e);
                        }
                    }
                });
            }

        }//Specify the topic name, key value, msg

        logger.info("This is the end");//When the reading is complete, inform logger
    }

    public Client tweetclient(BlockingQueue<String> msgQueue) {

        Hosts hosebirdHosts = new HttpHosts(Constants.STREAM_HOST);
        StatusesFilterEndpoint hosebirdEndpoint = new StatusesFilterEndpoint();
        List<String> terms = Lists.newArrayList("India ");//describe 
//anything for which we want to read the tweets.
        hosebirdEndpoint.trackTerms(terms);
        Authentication hosebirdAuth = new        OAuth1(consumerKey,consumerSecret,token,secret);
        ClientBuilder builder = new ClientBuilder()
                .name("Hosebird-Client-01")     // optional: mainly for the logs
                .hosts(hosebirdHosts)
                .authentication(hosebirdAuth)
                .endpoint(hosebirdEndpoint)
                .processor(new StringDelimitedProcessor(msgQueue));


        Client hosebirdClient = builder.build();
        return hosebirdClient; // Attempts to establish a connection.
 }
    public KafkaProducer<String,String> createKafkaProducer(){
 //creating kafka producer   
//creating producer properties   
        String bootstrapServers="127.0.0.1:9092";
        Properties properties= new Properties();
        properties.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,	   bootstrapServers);
        properties.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        KafkaProducer<String,String> first_producer = new KafkaProducer<String, String>(properties);
        return first_producer;

    }
}

在上面的代码中，用户将指定 consumerKey、consumerSecret 密钥、token 密钥以及密钥。由于它是敏感信息，因此无法显示。从 'developer.twitter.com' 复制密钥并粘贴到各自的位置。