我正在Kafka推特上试用下面教程中的示例代码。
https://www.tutorialspoint.com/apache_kafka/apache_kafka_real_time_application.htm
这是我的密码:
import java.util.Arrays;
import java.util.Properties;
import java.util.concurrent.LinkedBlockingQueue;
import twitter4j.*;
import twitter4j.conf.*;
import twitter4j.StatusListener;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
public class KafkaTwitterProducer {
public static void main(String[] args) throws Exception {
LinkedBlockingQueue<Status> queue = new LinkedBlockingQueue<Status>(1000);
String consumerKey = “XXXXXXXXXXXXXXXXX”; //args[0].toString();
String consumerSecret = "XXXXXXXXXXXXXXXXX"; //args[1].toString();
String accessToken = "XXXXXXXXXXXXXXXXX" ; //args[2].toString();
String accessTokenSecret = "XXXXXXXXXXXXXXXXX" ; //args[3].toString();
String topicName = "twittertest" ; //args[4].toString();
//String[] arguments = args.clone();
String[] keyWords = {“Hello”,”Hi”,”Welcome”}; //Arrays.copyOfRange(arguments, 5, arguments.length);
ConfigurationBuilder cb = new ConfigurationBuilder();
cb.setDebugEnabled(true)
.setOAuthConsumerKey(consumerKey)
.setOAuthConsumerSecret(consumerSecret)
.setOAuthAccessToken(accessToken)
.setOAuthAccessTokenSecret(accessTokenSecret);
TwitterStream twitterStream = new TwitterStreamFactory(cb.build()).getInstance();
StatusListener listener = new StatusListener() {
@Override
public void onStatus(Status status) {
queue.offer(status);
System.out.println("@" + status.getUser().getScreenName()
+ " - " + status.getText());
// System.out.println("@" + status.getUser().getScreen-Name());
/*for(URLEntity urle : status.getURLEntities()) {
System.out.println(urle.getDisplayURL());
}*/
/*for(HashtagEntity hashtage : status.getHashtagEntities()) {
System.out.println(hashtage.getText());
}*/
}
@Override
public void onDeletionNotice(StatusDeletionNotice statusDeletionNotice) {
System.out.println("Got a status deletion notice id:"
+ statusDeletionNotice.getStatusId());
}
@Override
public void onTrackLimitationNotice(int numberOfLimitedStatuses) {
System.out.println("Got track limitation notice:" +
numberOfLimitedStatuses);
}
@Override
public void onScrubGeo(long userId, long upToStatusId) {
System.out.println("Got scrub_geo event userId:" + userId +
"upToStatusId:" + upToStatusId);
}
@Override
public void onStallWarning(StallWarning warning) {
// System.out.println("Got stall warning:" + warning);
}
@Override
public void onException(Exception ex) {
ex.printStackTrace();
}
};
twitterStream.addListener(listener);
FilterQuery query = new FilterQuery().track(keyWords);
twitterStream.filter(query);
Thread.sleep(5000);
//Add Kafka producer config settings
Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("client.id", "SampleProducer");
props.put("auto.commit.interval.ms", "1000");
props.put("session.timeout.ms", "30000");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
//props.put("key.serializer",
// "org.apache.kafka.common.serialization.StringSerializer");
//props.put("value.serializer",
// "org.apache.kafka.common.serialization.StringSerializer");
Producer<String, String> producer = new KafkaProducer<String, String>(props);
int i = 0;
int j = 0;
while(i < 10) {
Status ret = queue.poll();
if (ret == null) {
Thread.sleep(100);
i++;
}else {
for(HashtagEntity hashtage : ret.getHashtagEntities()) {
System.out.println("Hashtag: " + hashtage.getText());
producer.send(new ProducerRecord<String, String>(
topicName, Integer.toString(j++), hashtage.getText()));
}
}
}
producer.close();
Thread.sleep(5000);
twitterStream.shutdown();
}
}
当我把它作为java应用程序运行时,我得到了以下错误:(这不是编译/构建错误)
Read timed out
Relevant discussions can be found on the Internet at:
http://www.google.co.jp/search?q=1169356e or
http://www.google.co.jp/search?q=c04b39f0
TwitterException{exceptionCode=[1169356e-c04b39f0 c2863472-491bffd7], statusCode=-1, message=null, code=-1, retryAfter=-1, rateLimitStatus=null, version=4.0.4}
at twitter4j.HttpClientImpl.handleRequest(HttpClientImpl.java:179)
at twitter4j.HttpClientBase.request(HttpClientBase.java:57)
at twitter4j.HttpClientBase.post(HttpClientBase.java:86)
at twitter4j.TwitterStreamImpl.getFilterStream(TwitterStreamImpl.java:346)
at twitter4j.TwitterStreamImpl$8.getStream(TwitterStreamImpl.java:322)
at twitter4j.TwitterStreamImpl$TwitterStreamConsumer.run(TwitterStreamImpl.java:552)
Caused by: java.net.SocketTimeoutException: Read timed out
at java.net.SocketInputStream.socketRead0(Native Method)
at java.net.SocketInputStream.socketRead(SocketInputStream.java:116)
at java.net.SocketInputStream.read(SocketInputStream.java:170)
at java.net.SocketInputStream.read(SocketInputStream.java:141)
at sun.security.ssl.InputRecord.readFully(InputRecord.java:465)
at sun.security.ssl.InputRecord.read(InputRecord.java:503)
at sun.security.ssl.SSLSocketImpl.readRecord(SSLSocketImpl.java:973)
at sun.security.ssl.SSLSocketImpl.readDataRecord(SSLSocketImpl.java:930)
at sun.security.ssl.AppInputStream.read(AppInputStream.java:105)
at java.io.BufferedInputStream.fill(BufferedInputStream.java:246)
at java.io.BufferedInputStream.read1(BufferedInputStream.java:286)
at java.io.BufferedInputStream.read(BufferedInputStream.java:345)
at sun.net.www.http.HttpClient.parseHTTPHeader(HttpClient.java:704)
at sun.net.www.http.HttpClient.parseHTTP(HttpClient.java:647)
at sun.net.www.protocol.http.HttpURLConnection.getInputStream0(HttpURLConnection.java:1536)
at sun.net.www.protocol.http.HttpURLConnection.getInputStream(HttpURLConnection.java:1441)
at java.net.HttpURLConnection.getResponseCode(HttpURLConnection.java:480)
at sun.net.www.protocol.https.HttpsURLConnectionImpl.getResponseCode(HttpsURLConnectionImpl.java:338)
at twitter4j.HttpResponseImpl.<init>(HttpResponseImpl.java:35)
at twitter4j.HttpClientImpl.handleRequest(HttpClientImpl.java:143)
... 5 more
我不确定这里有什么问题。有人能给我建议解决办法吗?
确定更新:如果关键字是通用的,比如string[]keywords={“usa”,“basketball”,“sports};
如果我用公司名称、产品名称等特定关键字将此更改为我的需求,例如:string[]keywords={“”,“yyy”,“zzz”};然后java应用程序将被终止。原因是什么?如何在这个代码中修复它?请告知?
1条答案
按热度按时间wpcxdonn1#
twitter4j源代码显示,抛出此异常是因为http连接超时。
通过设置连接超时的低值,我得到了类似的异常。
这是我得到的堆栈跟踪。
对于您的示例,请尝试为httpstreamingreadtimeout设置更高的值。代码中的默认值是40秒。尝试将其设置为120000(毫秒)或更高。这应该管用。