我正在实现一个集成了Kafka,Elasticsearch和Kibana的微服务项目。我已经为容器之间的通信配置了网络,但是我无法让数据收集微服务与版本为7.6.2的运行ElasticSearch oss镜像的容器进行通信。
启动Spring Boot微服务时,我可以看到以下日志:
INFO 1 --- [ main] o.s.d.elasticsearch.support.VersionInfo : Version Spring Data Elasticsearch: 4.0.2.RELEASE
INFO 1 --- [ main] o.s.d.elasticsearch.support.VersionInfo : Version Elasticsearch Client in build: 7.6.2
INFO 1 --- [ main] o.s.d.elasticsearch.support.VersionInfo : Version Elasticsearch Client used: 7.6.2
WARN 1 --- [ main] .d.e.r.s.AbstractElasticsearchRepository : Cannot create index: Connection refused; nested exception is java.lang.RuntimeException: Connection refused
以下是ElasticSearch容器配置:elasticsearch:
image: docker.elastic.co/elasticsearch/elasticsearch-oss:7.6.2
environment:
- discovery.type=single-node
- cluster.name=covid-tweets-es-cluster
- bootstrap.memory_lock=true
- "ES_JAVA_OPTS=-Xms512m -Xmx512m"
ulimits:
memlock:
soft: -1
hard: -1
volumes:
- elasticsearch-data:/usr/share/elasticsearch/data
ports:
- 9300:9300
- 9200:9200
networks:
- covid_processor_network
这是微服务的Spring Data ElasticSearch配置:## Kafka Binder Props
spring.cloud.stream.kafka.binder.brokers: kafka:9092
spring.cloud.stream.kafka.binder.auto-create-topics: false
spring.cloud.stream.kafka.binder.configuration.auto.offset.reset: latest
spring.cloud.stream.bindings.processed-tweets.group: tweets-collector
## Persistence
spring.data.elasticsearch.repositories.enabled: true
spring.data.elasticsearch.cluster-nodes: elasticsearch:9300
spring.data.elasticsearch.cluster-name: covid-tweets-es-cluster
spring.data.elasticsearch.rest.uris: elasticsearch:9200
从微服务容器中,我可以连接到elasticsearch容器这是客户端代码,非常简单,仅使用实现ElasticsearchRepository的TweetReopistory存储库
@StreamListener(AppStreamsConfig.PROCESSED_TWEETS_CHANNEL)
public void onNewProcessedTweet(
@Payload final TweetDTO newProcessedTweet,
@Header(KafkaHeaders.RECEIVED_TOPIC) String topic,
@Header(KafkaHeaders.RECEIVED_PARTITION_ID) Integer partition,
@Header(KafkaHeaders.OFFSET) Long offset,
@Header(IntegrationMessageHeaderAccessor.DELIVERY_ATTEMPT) Integer deliveryAttempt) {
log.info("NewsProcessedTweet with id '{}' and text '{}' received from bus. topic: {}, partition: {}, offset: {}, deliveryAttempt: {}",
newProcessedTweet.getId(), newProcessedTweet.getText(), topic, partition, offset, deliveryAttempt);
try {
tweetService.save(newProcessedTweet);
} catch (final Exception ex) {
ex.printStackTrace();
log.error("Collect Tweet Exception -> " + ex.getMessage());
}
}
@Service("tweetsService")
@RequiredArgsConstructor
public class TweetsServiceImpl implements ITweetsService {
/**
* Tweets Repository
*/
private final TweetsRepository tweetsRepository;
private final TweetDtoMapper tweetDtoMapper;
@Override
public void save(TweetDTO tweetDto) {
Assert.notNull(tweetDto, "Tweet can not be null");
final TweetEntity tweetToSave = tweetDtoMapper.dtoToEntity(tweetDto);
tweetsRepository.save(tweetToSave);
}
有人知道我在做什么错吗?谢谢
最佳答案
您正在使用哪个客户端? RestHighLevelClient还是TransportClient?
如果使用RestHighLevelClient,请尝试使用9200端口。
还要共享客户端使用代码以更好地理解。
关于java - AbstractElasticsearchRepository:无法创建索引:连接被拒绝;嵌套的异常是java.lang.RuntimeException:连接被拒绝,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/63330548/