Spring cloud stream实现Kafka的消息收发

用Spring cloud stream可以很方便的实现对Kafka消息的收发，以下是我按照Spring官网的例子实现的一个Kafka的应用。

这个例子是实现一个电信公司收集用户消费电信服务，并计算费用的场景。包括了三个应用程序。

1.记录用户使用电信服务时长：

这个应用将模拟生成用户的话单，包括了用户ID，语音呼叫时长，数据业务流量的信息，并把话单信息发送到Kafka。

用Spring boot创建一个名为usage-detail-sender-kafka的应用，依赖里面选择Spring for Apache Kafka, Spring cloud stream, Acutator

新建一个名为UsageDetail的类，代码如下：

package cn.roygao.usagedetailsender;
 
public class UsageDetail {
    private String userId;
    private long duration;
    private long data;
 
    public String getUserId() {
        return this.userId;
    }
 
    public void setUserId(String userId) {
        this.userId = userId;
    }
 
    public long getDuration() {
        return this.duration;
    }
 
    public void setDuration(long duration) {
        this.duration = duration;
    }
 
    public long getData() {
        return this.data;
    }
 
    public void setData(long data) {
        this.data = data;
    }
}

新建一个名为UsageDetailSender的类，实现发送Kafka消息的功能，代码如下：

package cn.roygao.usagedetailsender;
 
import java.util.Random;
import java.util.function.Supplier;
 
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
 
@Configuration
public class UsageDetailSender {
    private String[] users = {"user1", "user2", "user3", "user4", "user5"};
 
    @Bean
    public Supplier sendEvents() {
        return () -> {
            UsageDetail usageDetail = new UsageDetail();
            usageDetail.setUserId(this.users[new Random().nextInt(5)]);
            usageDetail.setDuration(new Random().nextInt(300));
            usageDetail.setData(new Random().nextInt(700));
            return usageDetail;
        };
    }
}

这个代码里面使用了函数式编程的模式，Supplier实现了一个发送Kafka消息的功能。

在application.properties里面添加以下配置：

spring.cloud.stream.function.bindings.sendEvents-out-0=output
spring.cloud.stream.bindings.output.destination=usage-detail
# Spring Boot will automatically assign an unused http port. This may be overridden using external configuration.
server.port=0

这个配置的第一行的意思是给这个Kafka的输出取一个别名，因为默认的这个Supplier的输出是绑定到spring.cloud.stream.function.bindings.xxxx-out-0的，其中xxxx代表这个输出函数的名字，在这里是sendEvents，因此我们用一个名为output的来代替xxxx-out-0这个名字。

在第二行配置就是设定这个输出的目的地消息主题是usage-detail

然后运行./mvnw clean package来编译。

2. 计算用户使用电信服务的费用：

这个应用将订阅Kafka的相关消息主题，接收应用1生成的用户的话单，然后计算相关的费用，并把费用发送到另外一个Kafka的消息主题。

用Spring boot创建一个名为usage-cost-processor-kafka的应用，依赖里面选择Spring for Apache Kafka, Spring cloud stream, Acutator

新建一个名为UsageDetail的类，如以上应用1的代码。

新建一个名为UsageCostDetail的类，代码如下：

package cn.roygao.usagecostprocessor;
 
public class UsageCostDetail {
    private String userId;
    private double callCost;
    private double dataCost;
 
 
    public String getUserId() {
        return this.userId;
    }
 
    public void setUserId(String userId) {
        this.userId = userId;
    }
 
    public double getCallCost() {
        return this.callCost;
    }
 
    public void setCallCost(double callCost) {
        this.callCost = callCost;
    }
 
    public double getDataCost() {
        return this.dataCost;
    }
 
    public void setDataCost(double dataCost) {
        this.dataCost = dataCost;
    }
 
    @Override
    public String toString() {
        return "{" +
            " userId='" + getUserId() + "'" +
            ", callCost='" + getCallCost() + "'" +
            ", dataCost='" + getDataCost() + "'" +
            "}";
    }
}

新建一个UsageCostProcessor的类，用于接收应用1发出的消息，计算费用之后再发消息到Kafka，代码如下：

package cn.roygao.usagecostprocessor;
 
import java.util.function.Function;
 
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
 
@Configuration
public class UsageCostProcessor {
    private double ratePerSecond = 0.1;
	private double ratePerMB = 0.05;
 
    @Bean
    public Function processUsageCost() {
		return usageDetail -> {
			UsageCostDetail usageCostDetail = new UsageCostDetail();
			usageCostDetail.setUserId(usageDetail.getUserId());
			usageCostDetail.setCallCost(usageDetail.getDuration() * this.ratePerSecond);
			usageCostDetail.setDataCost(usageDetail.getData() * this.ratePerMB);
			return usageCostDetail;
		};
	}
}

这里使用了Function来定义一个Kafka的接收和发送，其中第一个参数表示接收的消息，第二个参数表示发送的消息。

在application.properties里面添加以下配置：

spring.cloud.stream.function.bindings.processUsageCost-in-0=input
spring.cloud.stream.function.bindings.processUsageCost-out-0=output
spring.cloud.stream.bindings.input.destination=usage-detail
spring.cloud.stream.bindings.output.destination=usage-cost
# Spring Boot will automatically assign an unused http port. This may be overridden using external configuration.
server.port=0

3. 接收用户电信费用：

这个应用将订阅Kafka的主题，接收应用2发送的费用信息并在日志中打印出来。

新建一个名为usage-cost-logger-kafka的应用，依赖里面选择Spring for Apache Kafka, Spring cloud stream, Acutator

新建一个名为UsageCostDetail的类，如以上应用2的代码。

新建一个名为UsageCostLogger的类，代码如下：

package cn.roygao.usagecostlogger;
 
import java.util.function.Consumer;
 
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
 
@Configuration
public class UsageCostLogger {
    private static final Logger logger = LoggerFactory.getLogger(UsageCostLoggerKafkaApplication.class);
 
    @Bean
	public Consumer process() {
		return usageCostDetail -> {
			logger.info(usageCostDetail.toString());
		};
	}
}

这里使用了Consumer来实现对Kafka消息的消费。

在application.properties里面添加以下的配置：

spring.cloud.stream.function.bindings.process-in-0=input
spring.cloud.stream.bindings.input.destination=usage-cost
# Spring Boot will automatically assign an unused http port. This may be overridden using external configuration.
server.port=0

部署测试

以上三个应用编译好之后，我们就可以进行测试了。首先启动Kafka，这里我采用运行Docker的方式来启动Kafka，下载以下的docker-compose文件：

curl --silent --output docker-compose.yml \
  https://raw.githubusercontent.com/confluentinc/cp-all-in-one/6.1.0-post/cp-all-in-one/docker-compose.yml

然后运行docker compose up启动，之后在浏览器访问localhost:9021即可查看Kafka控制台的信息。

分别运行以上编译好的应用，然后在Kafka的控制台里面可以看到有usage-detail, usage-cost这两个主题的信息，并且在第三个应用输出的日志中可以看到打印出来的用户费用的信息。

以上就是一个简单的应用Spring cloud stream编写Kafka应用的例子。