Spring Cloud (Spring Cloud Stream)解析
执行脚本目录 /bin
windows 在其单独的目录
快速上手
下载并且解压kafka压缩包
运行服务
以Windows为例,首先打开cmd:
1. 启动zookeeper:
bin\windows\zookeeper-server-start.bat config\zookeeper.properties
2. 启动kafka:
bin\windows\kafka-server-start.bat config\server.properties
创建主题topic
bin\windows\kafka-topics.bat --create --bootstrap-server localhost:9092 --replication-factor 1 --partitions 1 --topic cloud
cloud
生产者:发送消息
\bin\windows>kafka-console-producer.bat --broker-list localhost:9092 --topic cloud
>hello
消费者:接收消息
\bin\windows>kafka-console-consumer.bat --bootstrap-server localhost:9092 --topic cloud --from-beginning
hello
同类产品比较
ActiveMQ:JMS(Java Message Service)规范实现
RabbitMQ:AMQP(Advanced Message Queue Protocol)规范实现
Kafka:并非某种规范实现,它的灵活和性能相对是优势
Spring Kafka
import java.util.Properties;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.serialization.StringSerializer;
public class TestKafka{
public static void main(String[] args) throws ExecutionException, InterruptedException {
Properties properties = new Properties();
properties.setProperty("bootstrap.servers", "localhost:9092");
properties.setProperty("key.serializer", StringSerializer.class.getName());
properties.setProperty("value.serializer", StringSerializer.class.getName());
// 创建 Kafka Producer
KafkaProducer<String, String> kafkaProducer = new KafkaProducer(properties);
// 创建Kafka消息 = ProducerRecord
String topic = "cloud";
Integer partition = 0;
Long timestamp = System.currentTimeMillis();
String key = "message-key";
String value = "how are you!";
ProducerRecord<String, String> record = new ProducerRecord<String, String>(topic, partition, timestamp, key, value);
// 发送kafka消息
Future<RecordMetadata> metadataFuture = kafkaProducer.send(record);
// 强制执行
metadataFuture.get();
}
}
设计模式
Spring社区对data(Spring-data)操作,有一个基本的模式,Template模式:
JDBC:JdbcTemplate
Redis:RedisTemplate
Kafka:KafkaTemplate
JMS:JmsTemplate
Rest:RestTemplate
XXXTemplate一定实现XXXOpeations
KafkaTemplate implements KafkaOpeations
Maven依赖
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka</artifactId>
</dependency>
自动装配器: 外汇返佣http://www.fx61.com/,KafkaAutoConfiguration
其中KafkaTemplate会被自动装配:
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;
import org.springframework.kafka.support.ProducerListener;
@Configuration
public class KafkaAutoConfiguration {
private final KafkaProperties properties;
public KafkaAutoConfiguration(KafkaProperties properties) {
this.properties = properties;
}
@Bean
@ConditionalOnMissingBean(KafkaTemplate.class)
public KafkaTemplate<?,?> kafkaTemplate(ProducerFactory<Object,Object> kafkaProducerFactory,
ProducerListener<Object,Object> kafkaProducerListener){
KafkaTemplate<Object,Object> kafkaTemplate = new KafkaTemplate<Object,Object>(kafkaProducerFactory);
kafkaTemplate.setProducerListener(kafkaProducerListener);
kafkaTemplate.setDefaultTopic(this.properties.getTemplate().getDefaultTopic());
return kafkaTemplate;
}
}
创建生产者
增加生产者配置
application.properties
全局配置:
### Kafka生产者配置
spring.kafka.producer.bootstrapServers = localhost:9092
### Kafka生产者配置
# spring.kafka.producer.bootstrapServers = localhost:9092
spring.kafka.producer.keySerializer = org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.valueSerializer = org.apache.kafka.common.serialization.StringSerializer
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;
@RestController
public class KafkaProducerController {
private final KafkaTemplate<String, String> kafkaTemplate;
private final String topic;
@Autowired
public KafkaProducerController(KafkaTemplate<String, String> kafkaTemplate, @Value("${kafka.topic}") String topic) {
this.kafkaTemplate = kafkaTemplate;
this.topic = topic;
}
@PostMapping("/message/send")
public Boolean sendMessage(@RequestParam(required=false)String message) {
kafkaTemplate.send(topic, message);
return true;
}
}
创建消费者
增加消费者配置
### Kafka消费者配置
spring.kafka.consumer.groupId = cloud-1
spring.kafka.consumer.keyDeserializer = org.apache.kafka.common.serialization.StringDeserializer
spring.kafka.consumer.valueDeserializer = org.apache.kafka.common.serialization.StringDeserializer
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
@Component
public class KafkaConsumerListener {
@KafkaListener(topics = "${kafka.topic}")
public void onMessage(String message) {
System.out.print("kafka 消费者监听器,接收到消息:" + message);
}
}
端口信息
spring-cloud-zuul:7070
person-client:8080
person-service:9090
Eureka Server:12345
ZipKin Server:23456
Config Server:10001
服务启动顺序
zipkin Server
Eureka Server
spring-cloud-config-server
person-server
person-client
spring-cloud-zuul
spring-cloud-sleuth
spring-cloud-sleuth-demo改造
增加Eureka客户端依赖
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-starter-netflix-eureka-client</artifactId>
</dependency>
配置调整
spring.application.name = spring-cloud-sleuth
server.port = 6060
spring.zipkin.base-url=http://localhost:23456/
eureka.client.serviceUrl.defaultZone=http://localhost:12345/eureka
调整代码链接:spring-cloud-zuul
完整调用链路
spring-cloud-sleuth → spring-cloud-zuul → person-client → person-service
@RestController
public class TestLoggerController {
final static Logger LOGGER = LoggerFactory.getLogger(TestLoggerController.class);
@Autowired
@Qualifier("restTemplate")
private RestTemplate restTemplate;
@GetMapping("/send")
public void send() {
LOGGER.info(" 欢迎欢迎!");
}
@GetMapping("/to/zuul/pseron-clint/findall")
public Object findall() {
LOGGER.info("TestLoggerController#findall()");
return restTemplate.getForObject("http://spring-cloud-zuul/person-client/person/findall", Object.class);
}
最新文章
- 如何实现 javascript “同步”调用 app 代码
- Swift2.1 语法指南——错误处理
- HDU 4950 Monster
- UISegmentedControl的使用
- iOS - UIColor
- php redis 分布式类
- Vmware出现报错The VMware Authorization Service is not running.之后无法上网解决
- Qt: 访问容器(三种方法,加上for循环就四种了)good
- 20160326 javaweb 请求转发和请求包含
- Linux SSH 互信
- Hibernate入门之配置文件
- 如何在Eclipse配置Tomcat服务器
- Java中的Math类的简单实用
- 编译安装Keepalived2.0.0
- 在python3中安装mysql扩展,No module named &#39;ConfigParser&#39;
- IdentityServer4 中文文档 -13- (快速入门)切换到混合流并添加 API 访问
- Struts2之类型转换
- 第二个Sprint
- asp.net mvc controller调用js
- OC开发_Storyboard——视图控制生命周期以及NSNotifications