未记录事实核验 · 状态：维护中
记录仅说明核验时间与维护状态，不保证适用于所有环境；使用前请留意正文中的版本、适用条件与风险提示。
warning 适用范围与待维护事项
本文适用于具备Java、分布式计算和Kafka消费语义基础的工程师进行示例代码审阅，不是完整可部署项目。原稿未指定框架版本、构建依赖和集群条件，多处实体、导入及方法省略；页面统计未实现Parquet落盘，Kafka消费者失败处理仅记录日志，超时任务与后续消费还可能交错并越过未成功处理的偏移量。Flink、CEP、Kafka Streams、YARN管理和集群部署仅在概览中列出，正文未展开。上线前须补齐材料并验证故障、重放和提交语义；会员分级不代表代码已达到生产质量。
Java大数据技术栈 - Hadoop/Spark/Kafka企业级开发实战
abstract 📋 Java大数据技术栈概览
🔍 为什么Java是大数据开发的主流语言？ Java在大数据生态系统中占据核心地位，主流的大数据框架如Hadoop、Spark、Kafka、Flink等都是基于Java或Scala（运行在JVM上）开发的。掌握Java大数据技术栈是数据开发工程师的必备技能。
🎯 本教程将帮助你掌握：
- 🐘 Hadoop生态开发：HDFS操作、MapReduce编程、YARN资源管理
- ⚡ Spark计算引擎：RDD编程、DataFrame/Dataset API、Spark SQL
- 🔄 Kafka消息系统：生产者消费者开发、Streams流处理
- 🌊 Flink实时计算：DataStream API、CEP复杂事件处理
- 🏗️ 企业级实践：集群部署、性能调优、监控运维
⏱️ 预计学习时长：180分钟用于阅读和梳理本文示例；实际练习、环境搭建与工程能力掌握所需时间因基础而异
🐘 Hadoop生态系统开发
📁 HDFS Java API深度应用
HDFS作为Hadoop分布式文件系统，是大数据存储的基础：
/**
* HDFS文件操作管理器
* 提供企业级的HDFS文件操作封装
*/
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;
import org.apache.hadoop.io.IOUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class HDFSManager {
private static final Logger logger = LoggerFactory.getLogger(HDFSManager.class);
private final FileSystem fileSystem;
private final Configuration configuration;
public HDFSManager(String hdfsUri) throws IOException {
this.configuration = new Configuration();
// HDFS配置优化
configuration.set("fs.defaultFS", hdfsUri);
configuration.set("dfs.client.block.write.replace-datanode-on-failure.enable", "true");
configuration.set("dfs.client.block.write.replace-datanode-on-failure.policy", "NEVER");
configuration.setInt("dfs.replication", 3);
configuration.setLong("dfs.block.size", 128 * 1024 * 1024); // 128MB
this.fileSystem = FileSystem.get(configuration);
logger.info("HDFS连接初始化完成: {}", hdfsUri);
}
/**
* 上传本地文件到HDFS
*/
public boolean uploadFile(String localPath, String hdfsPath, boolean overwrite) {
try {
Path localFilePath = new Path(localPath);
Path hdfsFilePath = new Path(hdfsPath);
// 检查本地文件是否存在
File localFile = new File(localPath);
if (!localFile.exists()) {
logger.error("本地文件不存在: {}", localPath);
return false;
}
// 创建HDFS目录
Path parentDir = hdfsFilePath.getParent();
if (!fileSystem.exists(parentDir)) {
fileSystem.mkdirs(parentDir);
logger.info("创建HDFS目录: {}", parentDir);
}
// 上传文件
long startTime = System.currentTimeMillis();
fileSystem.copyFromLocalFile(false, overwrite, localFilePath, hdfsFilePath);
long endTime = System.currentTimeMillis();
// 验证上传结果
if (fileSystem.exists(hdfsFilePath)) {
FileStatus fileStatus = fileSystem.getFileStatus(hdfsFilePath);
logger.info("文件上传成功: {} -> {}, 大小: {} bytes, 耗时: {} ms",
localPath, hdfsPath, fileStatus.getLen(), (endTime - startTime));
return true;
} else {
logger.error("文件上传失败: {}", hdfsPath);
return false;
}
} catch (IOException e) {
logger.error("文件上传异常: {}", e.getMessage(), e);
return false;
}
}
/**
* 从HDFS下载文件到本地
*/
public boolean downloadFile(String hdfsPath, String localPath) {
try {
Path hdfsFilePath = new Path(hdfsPath);
Path localFilePath = new Path(localPath);
// 检查HDFS文件是否存在
if (!fileSystem.exists(hdfsFilePath)) {
logger.error("HDFS文件不存在: {}", hdfsPath);
return false;
}
// 创建本地目录
File localFile = new File(localPath);
File parentDir = localFile.getParentFile();
if (parentDir != null && !parentDir.exists()) {
parentDir.mkdirs();
logger.info("创建本地目录: {}", parentDir.getAbsolutePath());
}
// 下载文件
long startTime = System.currentTimeMillis();
fileSystem.copyToLocalFile(false, hdfsFilePath, localFilePath, true);
long endTime = System.currentTimeMillis();
logger.info("文件下载成功: {} -> {}, 耗时: {} ms",
hdfsPath, localPath, (endTime - startTime));
return true;
} catch (IOException e) {
logger.error("文件下载异常: {}", e.getMessage(), e);
return false;
}
}
/**
* 读取HDFS文件内容
*/
public String readFile(String hdfsPath) throws IOException {
Path filePath = new Path(hdfsPath);
if (!fileSystem.exists(filePath)) {
throw new FileNotFoundException("HDFS文件不存在: " + hdfsPath);
}
StringBuilder content = new StringBuilder();
try (FSDataInputStream inputStream = fileSystem.open(filePath);
BufferedReader reader = new BufferedReader(new InputStreamReader(inputStream, StandardCharsets.UTF_8))) {
String line;
while ((line = reader.readLine()) != null) {
content.append(line).append("\n");
}
}
logger.debug("读取HDFS文件成功: {}, 内容长度: {} 字符", hdfsPath, content.length());
return content.toString();
}
/**
* 写入内容到HDFS文件
*/
public boolean writeFile(String hdfsPath, String content, boolean overwrite) {
try {
Path filePath = new Path(hdfsPath);
// 检查文件是否存在
if (fileSystem.exists(filePath) && !overwrite) {
logger.warn("HDFS文件已存在且不允许覆盖: {}", hdfsPath);
return false;
}
// 创建目录
Path parentDir = filePath.getParent();
if (!fileSystem.exists(parentDir)) {
fileSystem.mkdirs(parentDir);
}
// 写入文件
try (FSDataOutputStream outputStream = fileSystem.create(filePath, overwrite);
BufferedWriter writer = new BufferedWriter(new OutputStreamWriter(outputStream, StandardCharsets.UTF_8))) {
writer.write(content);
writer.flush();
}
logger.info("写入HDFS文件成功: {}, 内容长度: {} 字符", hdfsPath, content.length());
return true;
} catch (IOException e) {
logger.error("写入HDFS文件异常: {}", e.getMessage(), e);
return false;
}
}
/**
* 列出目录下的文件和子目录
*/
public List<HDFSFileInfo> listFiles(String dirPath, boolean recursive) throws IOException {
Path path = new Path(dirPath);
List<HDFSFileInfo> fileInfos = new ArrayList<>();
if (!fileSystem.exists(path)) {
logger.warn("HDFS目录不存在: {}", dirPath);
return fileInfos;
}
FileStatus[] fileStatuses;
if (recursive) {
RemoteIterator<LocatedFileStatus> iterator = fileSystem.listFiles(path, true);
List<FileStatus> statusList = new ArrayList<>();
while (iterator.hasNext()) {
statusList.add(iterator.next());
}
fileStatuses = statusList.toArray(new FileStatus[0]);
} else {
fileStatuses = fileSystem.listStatus(path);
}
for (FileStatus status : fileStatuses) {
HDFSFileInfo fileInfo = new HDFSFileInfo();
fileInfo.setPath(status.getPath().toString());
fileInfo.setIsDirectory(status.isDirectory());
fileInfo.setSize(status.getLen());
fileInfo.setReplication(status.getReplication());
fileInfo.setBlockSize(status.getBlockSize());
fileInfo.setModificationTime(new Date(status.getModificationTime()));
fileInfo.setOwner(status.getOwner());
fileInfo.setGroup(status.getGroup());
fileInfo.setPermission(status.getPermission().toString());
fileInfos.add(fileInfo);
}
logger.debug("列出HDFS目录文件: {}, 文件数量: {}", dirPath, fileInfos.size());
return fileInfos;
}
/**
* 删除HDFS文件或目录
*/
public boolean delete(String hdfsPath, boolean recursive) {
try {
Path path = new Path(hdfsPath);
if (!fileSystem.exists(path)) {
logger.warn("HDFS路径不存在: {}", hdfsPath);
return false;
}
boolean result = fileSystem.delete(path, recursive);
if (result) {
logger.info("删除HDFS路径成功: {}", hdfsPath);
} else {
logger.error("删除HDFS路径失败: {}", hdfsPath);
}
return result;
} catch (IOException e) {
logger.error("删除HDFS路径异常: {}", e.getMessage(), e);
return false;
}
}
/**
* 关闭HDFS连接
*/
public void close() {
try {
if (fileSystem != null) {
fileSystem.close();
logger.info("HDFS连接已关闭");
}
} catch (IOException e) {
logger.error("关闭HDFS连接异常: {}", e.getMessage(), e);
}
}
}
/**
* HDFS文件信息实体
*/
public class HDFSFileInfo {
private String path;
private boolean isDirectory;
private long size;
private short replication;
private long blockSize;
private Date modificationTime;
private String owner;
private String group;
private String permission;
// getters and setters...
}
🔧 MapReduce作业开发
/**
* 词频统计MapReduce作业示例
* 展示企业级MapReduce开发的最佳实践
*/
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
/**
* 词频统计Mapper
*/
public class WordCountMapper extends Mapper<LongWritable, Text, Text, IntWritable> {
private static final Logger logger = LoggerFactory.getLogger(WordCountMapper.class);
private final IntWritable one = new IntWritable(1);
private final Text word = new Text();
// 停用词列表
private Set<String> stopWords;
@Override
protected void setup(Context context) throws IOException, InterruptedException {
super.setup(context);
// 初始化停用词列表
stopWords = new HashSet<>();
stopWords.addAll(Arrays.asList("the", "a", "an", "and", "or", "but", "in", "on", "at", "to", "for", "of", "with", "by"));
logger.info("WordCountMapper初始化完成，停用词数量: {}", stopWords.size());
}
@Override
protected void map(LongWritable key, Text value, Context context)
throws IOException, InterruptedException {
// 转换为小写并分割单词
String line = value.toString().toLowerCase();
String[] words = line.split("\\s+");
for (String wordStr : words) {
// 清理单词（移除标点符号等）
wordStr = wordStr.replaceAll("[^a-zA-Z0-9]", "");
if (!wordStr.isEmpty() && !stopWords.contains(wordStr) && wordStr.length() > 2) {
word.set(wordStr);
context.write(word, one);
}
}
}
}
/**
* 词频统计Reducer
*/
public class WordCountReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
private static final Logger logger = LoggerFactory.getLogger(WordCountReducer.class);
private final IntWritable result = new IntWritable();
@Override
protected void reduce(Text key, Iterable<IntWritable> values, Context context)
throws IOException, InterruptedException {
int sum = 0;
for (IntWritable value : values) {
sum += value.get();
}
// 只输出出现次数大于阈值的单词
Configuration conf = context.getConfiguration();
int threshold = conf.getInt("wordcount.threshold", 1);
if (sum >= threshold) {
result.set(sum);
context.write(key, result);
}
}
@Override
protected void cleanup(Context context) throws IOException, InterruptedException {
super.cleanup(context);
logger.info("WordCountReducer处理完成");
}
}
/**
* MapReduce作业驱动程序
*/
public class WordCountDriver {
private static final Logger logger = LoggerFactory.getLogger(WordCountDriver.class);
public static void main(String[] args) throws Exception {
if (args.length != 2) {
System.err.println("用法: WordCountDriver <input> <output>");
System.exit(-1);
}
Configuration conf = new Configuration();
// 作业配置优化
conf.setInt("mapreduce.task.io.sort.mb", 256); // Sort缓冲区大小
conf.setFloat("mapreduce.map.sort.spill.percent", 0.8f); // 溢出比例
conf.setInt("mapreduce.task.io.sort.factor", 64); // 合并因子
conf.setBoolean("mapreduce.map.output.compress", true); // 启用Map输出压缩
conf.set("mapreduce.map.output.compress.codec", "org.apache.hadoop.io.compress.SnappyCodec");
// 自定义参数
conf.setInt("wordcount.threshold", 5); // 词频阈值
Job job = Job.getInstance(conf, "word count");
job.setJarByClass(WordCountDriver.class);
// 设置Mapper和Reducer
job.setMapperClass(WordCountMapper.class);
// 不将带阈值过滤的WordCountReducer用作Combiner，避免提前丢弃局部计数
job.setReducerClass(WordCountReducer.class);
// 设置输出类型
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(IntWritable.class);
// 设置输入输出格式
job.setInputFormatClass(TextInputFormat.class);
job.setOutputFormatClass(TextOutputFormat.class);
// 设置输入输出路径
FileInputFormat.addInputPath(job, new Path(args[0]));
FileOutputFormat.setOutputPath(job, new Path(args[1]));
// 设置Reducer数量
job.setNumReduceTasks(2);
logger.info("提交WordCount作业");
long startTime = System.currentTimeMillis();
boolean success = job.waitForCompletion(true);
long endTime = System.currentTimeMillis();
logger.info("WordCount作业完成，耗时: {} 秒，结果: {}",
(endTime - startTime) / 1000, success ? "成功" : "失败");
if (success) {
// 输出作业统计信息
Counters counters = job.getCounters();
Counter mapInputRecords = counters.findCounter(TaskCounter.MAP_INPUT_RECORDS);
Counter reduceOutputRecords = counters.findCounter(TaskCounter.REDUCE_OUTPUT_RECORDS);
logger.info("输入记录数: {}", mapInputRecords.getValue());
logger.info("输出记录数: {}", reduceOutputRecords.getValue());
}
System.exit(success ? 0 : 1);
}
}
⚡ Spark计算引擎开发
🔥 Spark Core RDD编程
/**
* Spark RDD数据处理应用
* 展示企业级Spark RDD编程最佳实践
*/
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.*;
import org.apache.spark.storage.StorageLevel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import scala.Tuple2;
public class SparkDataProcessor {
private static final Logger logger = LoggerFactory.getLogger(SparkDataProcessor.class);
private final JavaSparkContext sparkContext;
public SparkDataProcessor(String appName, String master) {
SparkConf conf = new SparkConf()
.setAppName(appName)
.setMaster(master)
// Spark配置优化
.set("spark.sql.adaptive.enabled", "true")
.set("spark.sql.adaptive.coalescePartitions.enabled", "true")
.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
.set("spark.sql.execution.arrow.pyspark.enabled", "true")
.set("spark.dynamicAllocation.enabled", "true")
.set("spark.dynamicAllocation.minExecutors", "2")
.set("spark.dynamicAllocation.maxExecutors", "10");
this.sparkContext = new JavaSparkContext(conf);
logger.info("Spark应用启动: {}", appName);
}
/**
* 用户行为日志分析
*/
public void analyzeUserBehaviorLogs(String inputPath, String outputPath) {
logger.info("开始分析用户行为日志: {}", inputPath);
try {
// 读取日志文件
JavaRDD<String> logLines = sparkContext.textFile(inputPath);
// 解析日志数据
JavaRDD<UserBehavior> behaviors = logLines
.filter(line -> !line.trim().isEmpty()) // 过滤空行
.map(new ParseLogFunction())
.filter(Objects::nonNull) // 过滤解析失败的数据
.cache(); // 缓存中间结果
logger.info("有效日志记录数: {}", behaviors.count());
// 分析1: 用户活跃度统计
analyzeUserActivity(behaviors, outputPath + "/user_activity");
// 分析2: 页面访问统计
analyzePageViews(behaviors, outputPath + "/page_views");
// 分析3: 用户会话分析
analyzeUserSessions(behaviors, outputPath + "/user_sessions");
// 分析4: 异常行为检测
detectAnomalousActivity(behaviors, outputPath + "/anomalous_activity");
} catch (Exception e) {
logger.error("用户行为日志分析失败: {}", e.getMessage(), e);
throw new RuntimeException(e);
}
}
/**
* 用户活跃度分析
*/
private void analyzeUserActivity(JavaRDD<UserBehavior> behaviors, String outputPath) {
JavaPairRDD<String, Integer> userActivity = behaviors
.mapToPair(behavior -> new Tuple2<>(behavior.getUserId(), 1))
.reduceByKey((a, b) -> a + b)
.filter(tuple -> tuple._2() >= 5); // 过滤活跃度低的用户
// 按活跃度排序
JavaPairRDD<String, Integer> sortedActivity = userActivity
.mapToPair(tuple -> new Tuple2<>(tuple._2(), tuple._1()))
.sortByKey(false) // 降序排序
.mapToPair(tuple -> new Tuple2<>(tuple._2(), tuple._1()));
// 保存结果
sortedActivity
.map(tuple -> tuple._1() + "\t" + tuple._2())
.coalesce(1) // 合并为单个文件
.saveAsTextFile(outputPath);
logger.info("用户活跃度分析完成，活跃用户数: {}", userActivity.count());
}
/**
* 页面访问统计
*/
private void analyzePageViews(JavaRDD<UserBehavior> behaviors, String outputPath) {
JavaPairRDD<String, Long> pageViews = behaviors
.filter(behavior -> "page_view".equals(behavior.getAction()))
.mapToPair(behavior -> new Tuple2<>(behavior.getPage(), 1L))
.reduceByKey(Long::sum);
// 计算页面访问统计
JavaRDD<PageViewStats> pageStats = pageViews
.map(tuple -> {
PageViewStats stats = new PageViewStats();
stats.setPage(tuple._1());
stats.setViewCount(tuple._2());
return stats;
});
// 待补充：将pageStats转换为DataFrame并保存到outputPath下的Parquet文件；当前示例尚未实现
// 此处不停止共享Spark上下文；后续分析仍需使用它，全部分析结束后由close()关闭
logger.info("页面访问统计完成，页面数量: {}", pageViews.count());
}
/**
* 用户会话分析
*/
private void analyzeUserSessions(JavaRDD<UserBehavior> behaviors, String outputPath) {
// 按用户分组
JavaPairRDD<String, Iterable<UserBehavior>> userGroups = behaviors
.mapToPair(behavior -> new Tuple2<>(behavior.getUserId(), behavior))
.groupByKey();
// 会话分析
JavaRDD<UserSession> sessions = userGroups
.flatMap(new SessionAnalysisFunction());
// 会话统计
JavaPairRDD<String, SessionStats> sessionStats = sessions
.mapToPair(session -> new Tuple2<>(session.getUserId(),
new SessionStats(session.getDuration(), session.getPageCount())))
.reduceByKey((stats1, stats2) -> new SessionStats(
Math.max(stats1.getMaxDuration(), stats2.getMaxDuration()),
stats1.getTotalPages() + stats2.getTotalPages()
));
// 保存会话统计结果
sessionStats
.map(tuple -> String.format("%s\t%.2f\t%d",
tuple._1(),
tuple._2().getMaxDuration(),
tuple._2().getTotalPages()))
.saveAsTextFile(outputPath);
logger.info("用户会话分析完成，会话数量: {}", sessions.count());
}
/**
* 异常行为检测
*/
private void detectAnomalousActivity(JavaRDD<UserBehavior> behaviors, String outputPath) {
// 计算每小时的活动统计
JavaPairRDD<Integer, Long> hourlyActivity = behaviors
.mapToPair(behavior -> new Tuple2<>(behavior.getTimestamp().getHour(), 1L))
.reduceByKey(Long::sum);
// 计算平均活动水平
double avgActivity = hourlyActivity.values().collect().stream()
.mapToLong(Long::longValue)
.average()
.orElse(0.0);
// 检测异常活动（超过已统计小时平均活动量的3倍；此处未计算标准差）
double threshold = avgActivity * 3;
JavaPairRDD<Integer, Long> anomalousHours = hourlyActivity
.filter(tuple -> tuple._2() > threshold);
// 找出异常时段的用户行为
Set<Integer> anomalousHourSet = anomalousHours.keys().collect()
.stream().collect(Collectors.toSet());
JavaRDD<UserBehavior> anomalousActivity = behaviors
.filter(behavior -> anomalousHourSet.contains(behavior.getTimestamp().getHour()));
// 保存异常行为
anomalousActivity
.map(behavior -> String.format("%s\t%s\t%s\t%s",
behavior.getUserId(),
behavior.getAction(),
behavior.getPage(),
behavior.getTimestamp().toString()))
.saveAsTextFile(outputPath);
logger.info("异常行为检测完成，异常记录数: {}", anomalousActivity.count());
}
/**
* 关闭Spark上下文
*/
public void close() {
if (sparkContext != null) {
sparkContext.close();
logger.info("Spark应用已关闭");
}
}
}
/**
* 日志解析函数
*/
class ParseLogFunction implements Function<String, UserBehavior> {
private static final Logger logger = LoggerFactory.getLogger(ParseLogFunction.class);
@Override
public UserBehavior call(String logLine) throws Exception {
try {
// 假设日志格式: timestamp|userId|action|page|ip
String[] parts = logLine.split("\\|");
if (parts.length >= 5) {
UserBehavior behavior = new UserBehavior();
behavior.setTimestamp(LocalDateTime.parse(parts[0], DateTimeFormatter.ISO_LOCAL_DATE_TIME));
behavior.setUserId(parts[1]);
behavior.setAction(parts[2]);
behavior.setPage(parts[3]);
behavior.setIp(parts[4]);
return behavior;
}
} catch (Exception e) {
logger.warn("日志解析失败: {}", logLine);
}
return null;
}
}
/**
* 会话分析函数
*/
class SessionAnalysisFunction implements FlatMapFunction<Tuple2<String, Iterable<UserBehavior>>, UserSession> {
@Override
public Iterator<UserSession> call(Tuple2<String, Iterable<UserBehavior>> userGroup) throws Exception {
String userId = userGroup._1();
List<UserBehavior> behaviors = StreamSupport.stream(userGroup._2().spliterator(), false)
.sorted(Comparator.comparing(UserBehavior::getTimestamp))
.collect(Collectors.toList());
List<UserSession> sessions = new ArrayList<>();
if (behaviors.isEmpty()) {
return sessions.iterator();
}
// 会话切分逻辑（30分钟无活动则认为会话结束）
UserSession currentSession = new UserSession();
currentSession.setUserId(userId);
currentSession.setStartTime(behaviors.get(0).getTimestamp());
LocalDateTime lastActivity = behaviors.get(0).getTimestamp();
int pageCount = 1;
for (int i = 1; i < behaviors.size(); i++) {
UserBehavior behavior = behaviors.get(i);
// 检查会话间隔
long minutesBetween = ChronoUnit.MINUTES.between(lastActivity, behavior.getTimestamp());
if (minutesBetween > 30) { // 会话超时
// 结束当前会话
currentSession.setEndTime(lastActivity);
currentSession.setDuration(ChronoUnit.MINUTES.between(
currentSession.getStartTime(), currentSession.getEndTime()));
currentSession.setPageCount(pageCount);
sessions.add(currentSession);
// 开始新会话
currentSession = new UserSession();
currentSession.setUserId(userId);
currentSession.setStartTime(behavior.getTimestamp());
pageCount = 1;
} else {
pageCount++;
}
lastActivity = behavior.getTimestamp();
}
// 处理最后一个会话
currentSession.setEndTime(lastActivity);
currentSession.setDuration(ChronoUnit.MINUTES.between(
currentSession.getStartTime(), currentSession.getEndTime()));
currentSession.setPageCount(pageCount);
sessions.add(currentSession);
return sessions.iterator();
}
}
🔄 Kafka消息系统集成
📨 Kafka生产者开发
/**
* 高性能Kafka生产者
* 支持批量发送、异步处理、故障恢复
*/
import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class EnhancedKafkaProducer {
private static final Logger logger = LoggerFactory.getLogger(EnhancedKafkaProducer.class);
private final KafkaProducer<String, String> producer;
private final String topicName;
private final Counter successCounter;
private final Counter errorCounter;
private final Timer sendTimer;
public EnhancedKafkaProducer(String bootstrapServers, String topicName, MeterRegistry meterRegistry) {
this.topicName = topicName;
Properties props = new Properties();
// 基础配置
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
// 性能优化配置
props.put(ProducerConfig.ACKS_CONFIG, "all"); // 与下方显式启用幂等性的配置保持一致
props.put(ProducerConfig.RETRIES_CONFIG, 3); // 重试次数
props.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384); // 批处理大小
props.put(ProducerConfig.LINGER_MS_CONFIG, 5); // 等待时间
props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432); // 缓冲区大小
props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "snappy"); // 压缩算法
// 可靠性配置
props.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, 5);
props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true); // 幂等性
props.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG, 30000);
props.put(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG, 120000);
this.producer = new KafkaProducer<>(props);
// 监控指标
this.successCounter = Counter.builder("kafka.producer.success").register(meterRegistry);
this.errorCounter = Counter.builder("kafka.producer.error").register(meterRegistry);
this.sendTimer = Timer.builder("kafka.producer.send.time").register(meterRegistry);
logger.info("Kafka生产者初始化完成，主题: {}", topicName);
}
/**
* 同步发送消息
*/
public void sendSync(String key, String message) throws ExecutionException, InterruptedException {
Timer.Sample sample = Timer.start();
try {
ProducerRecord<String, String> record = new ProducerRecord<>(topicName, key, message);
RecordMetadata metadata = producer.send(record).get();
sample.stop(sendTimer);
successCounter.increment();
logger.debug("消息发送成功 - 主题: {}, 分区: {}, 偏移量: {}",
metadata.topic(), metadata.partition(), metadata.offset());
} catch (Exception e) {
sample.stop(sendTimer);
errorCounter.increment();
logger.error("同步发送消息失败: {}", e.getMessage(), e);
throw e;
}
}
/**
* 异步发送消息
*/
public void sendAsync(String key, String message, Callback callback) {
Timer.Sample sample = Timer.start();
ProducerRecord<String, String> record = new ProducerRecord<>(topicName, key, message);
producer.send(record, (metadata, exception) -> {
sample.stop(sendTimer);
if (exception == null) {
successCounter.increment();
logger.debug("异步消息发送成功 - 主题: {}, 分区: {}, 偏移量: {}",
metadata.topic(), metadata.partition(), metadata.offset());
} else {
errorCounter.increment();
logger.error("异步消息发送失败: {}", exception.getMessage(), exception);
}
if (callback != null) {
callback.onCompletion(metadata, exception);
}
});
}
/**
* 批量异步发送消息
*/
public void sendBatchAsync(List<MessageRecord> messages) {
if (messages.isEmpty()) {
return;
}
logger.info("开始批量发送消息，数量: {}", messages.size());
CountDownLatch latch = new CountDownLatch(messages.size());
AtomicInteger successCount = new AtomicInteger(0);
AtomicInteger errorCount = new AtomicInteger(0);
for (MessageRecord messageRecord : messages) {
sendAsync(messageRecord.getKey(), messageRecord.getMessage(), (metadata, exception) -> {
if (exception == null) {
successCount.incrementAndGet();
} else {
errorCount.incrementAndGet();
}
latch.countDown();
});
}
// 等待所有消息发送完成（带超时）
try {
boolean completed = latch.await(30, TimeUnit.SECONDS);
if (completed) {
logger.info("批量消息发送完成 - 成功: {}, 失败: {}",
successCount.get(), errorCount.get());
} else {
logger.warn("批量消息发送超时 - 已完成: {}, 总计: {}",
messages.size() - latch.getCount(), messages.size());
}
} catch (InterruptedException e) {
logger.error("批量消息发送被中断", e);
Thread.currentThread().interrupt();
}
}
/**
* 发送JSON格式的结构化数据
*/
public void sendJsonMessage(String key, Object data) {
try {
ObjectMapper mapper = new ObjectMapper();
String jsonMessage = mapper.writeValueAsString(data);
sendAsync(key, jsonMessage, null);
} catch (JsonProcessingException e) {
logger.error("JSON序列化失败: {}", e.getMessage(), e);
errorCounter.increment();
}
}
/**
* 发送用户行为事件
*/
public void sendUserEvent(UserEvent event) {
String key = event.getUserId();
// 添加元数据
event.setTimestamp(LocalDateTime.now());
event.setEventId(UUID.randomUUID().toString());
sendJsonMessage(key, event);
logger.debug("用户事件发送: userId={}, action={}",
event.getUserId(), event.getAction());
}
/**
* 获取生产者指标
*/
public ProducerMetrics getMetrics() {
Map<MetricName, ? extends Metric> metrics = producer.metrics();
ProducerMetrics producerMetrics = new ProducerMetrics();
for (Map.Entry<MetricName, ? extends Metric> entry : metrics.entrySet()) {
MetricName metricName = entry.getKey();
Metric metric = entry.getValue();
switch (metricName.name()) {
case "record-send-rate":
producerMetrics.setRecordSendRate((Double) metric.metricValue());
break;
case "record-error-rate":
producerMetrics.setRecordErrorRate((Double) metric.metricValue());
break;
case "request-latency-avg":
producerMetrics.setRequestLatencyAvg((Double) metric.metricValue());
break;
case "batch-size-avg":
producerMetrics.setBatchSizeAvg((Double) metric.metricValue());
break;
}
}
return producerMetrics;
}
/**
* 关闭生产者
*/
public void close() {
try {
producer.flush(); // 等待此前发送请求完成；是否成功仍须检查Future或回调结果
producer.close(Duration.ofSeconds(10));
logger.info("Kafka生产者已关闭");
} catch (Exception e) {
logger.error("关闭Kafka生产者异常: {}", e.getMessage(), e);
}
}
}
📥 Kafka消费者开发
/**
* 高性能Kafka消费者
* 支持多线程消费、偏移量管理、故障恢复
*/
public class EnhancedKafkaConsumer {
private static final Logger logger = LoggerFactory.getLogger(EnhancedKafkaConsumer.class);
private final KafkaConsumer<String, String> consumer;
private final List<String> topics;
private final MessageProcessor messageProcessor;
private final ExecutorService executorService;
private volatile boolean running = false;
// 监控指标
private final Counter consumedCounter;
private final Counter processedCounter;
private final Counter errorCounter;
private final Timer processTimer;
public EnhancedKafkaConsumer(String bootstrapServers, String groupId, List<String> topics,
MessageProcessor messageProcessor, MeterRegistry meterRegistry) {
this.topics = topics;
this.messageProcessor = messageProcessor;
Properties props = new Properties();
// 基础配置
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
// 消费策略配置
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); // 无有效已提交偏移量时，从最早可用偏移量开始消费
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false); // 手动提交偏移量
props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 500); // 每次poll的最大记录数
props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, 300000); // poll间隔
// 性能优化配置
props.put(ConsumerConfig.FETCH_MIN_BYTES_CONFIG, 1024); // 最小fetch字节数
props.put(ConsumerConfig.FETCH_MAX_WAIT_MS_CONFIG, 500); // 最大等待时间
props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 30000); // 会话超时
props.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, 3000); // 心跳间隔
this.consumer = new KafkaConsumer<>(props);
// 创建线程池用于消息处理
int threadCount = Math.max(2, Runtime.getRuntime().availableProcessors());
this.executorService = Executors.newFixedThreadPool(threadCount);
// 监控指标
this.consumedCounter = Counter.builder("kafka.consumer.consumed").register(meterRegistry);
this.processedCounter = Counter.builder("kafka.consumer.processed").register(meterRegistry);
this.errorCounter = Counter.builder("kafka.consumer.error").register(meterRegistry);
this.processTimer = Timer.builder("kafka.consumer.process.time").register(meterRegistry);
logger.info("Kafka消费者初始化完成，组ID: {}, 主题: {}, 处理线程数: {}",
groupId, topics, threadCount);
}
/**
* 启动消费者
*/
public void start() {
running = true;
consumer.subscribe(topics);
logger.info("开始消费Kafka消息，主题: {}", topics);
try {
while (running) {
ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
if (!records.isEmpty()) {
logger.debug("接收到 {} 条消息", records.count());
consumedCounter.increment(records.count());
processRecords(records);
}
}
} catch (WakeupException e) {
logger.info("消费者收到停止信号");
} catch (Exception e) {
logger.error("消费者运行异常: {}", e.getMessage(), e);
errorCounter.increment();
} finally {
try {
consumer.close();
logger.info("Kafka消费者已关闭");
} catch (Exception e) {
logger.error("关闭消费者异常: {}", e.getMessage(), e);
}
}
}
/**
* 处理消息记录
*/
private void processRecords(ConsumerRecords<String, String> records) {
// 按分区分组，在单批次任务内顺序处理；超时后跨批任务仍可能交错，不能保证跨批顺序
Map<TopicPartition, List<ConsumerRecord<String, String>>> partitionRecords = new HashMap<>();
for (ConsumerRecord<String, String> record : records) {
TopicPartition partition = new TopicPartition(record.topic(), record.partition());
partitionRecords.computeIfAbsent(partition, k -> new ArrayList<>()).add(record);
}
// 并行处理各个分区
List<CompletableFuture<Void>> futures = new ArrayList<>();
for (Map.Entry<TopicPartition, List<ConsumerRecord<String, String>>> entry : partitionRecords.entrySet()) {
TopicPartition partition = entry.getKey();
List<ConsumerRecord<String, String>> partitionMessages = entry.getValue();
CompletableFuture<Void> future = CompletableFuture.runAsync(() ->
processPartitionMessages(partition, partitionMessages), executorService);
futures.add(future);
}
// 等待所有分区处理完成
try {
CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]))
.get(30, TimeUnit.SECONDS);
// 手动提交偏移量
consumer.commitSync();
logger.debug("偏移量提交成功，处理消息数: {}", records.count());
} catch (TimeoutException e) {
logger.error("消息处理超时", e);
errorCounter.increment();
} catch (Exception e) {
logger.error("消息处理失败: {}", e.getMessage(), e);
errorCounter.increment();
}
}
/**
* 处理单个分区的消息
*/
private void processPartitionMessages(TopicPartition partition, List<ConsumerRecord<String, String>> messages) {
logger.debug("处理分区 {} 的 {} 条消息", partition, messages.size());
for (ConsumerRecord<String, String> record : messages) {
Timer.Sample sample = Timer.start();
try {
// 消息处理
ProcessingResult result = messageProcessor.process(
record.key(),
record.value(),
new MessageMetadata(record.topic(), record.partition(), record.offset(), record.timestamp())
);
sample.stop(processTimer);
if (result.isSuccess()) {
processedCounter.increment();
logger.debug("消息处理成功 - Key: {}, Offset: {}", record.key(), record.offset());
} else {
logger.warn("消息处理失败 - Key: {}, Offset: {}, 错误: {}",
record.key(), record.offset(), result.getErrorMessage());
errorCounter.increment();
// 错误处理策略
handleProcessingError(record, result);
}
} catch (Exception e) {
sample.stop(processTimer);
logger.error("消息处理异常 - Key: {}, Offset: {}, 异常: {}",
record.key(), record.offset(), e.getMessage(), e);
errorCounter.increment();
// 异常处理策略
handleProcessingException(record, e);
}
}
}
/**
* 处理错误消息
*/
private void handleProcessingError(ConsumerRecord<String, String> record, ProcessingResult result) {
// 可以选择：
// 1. 发送到死信队列
// 2. 记录到错误日志
// 3. 重试处理
logger.warn("错误消息处理策略 - Topic: {}, Partition: {}, Offset: {}",
record.topic(), record.partition(), record.offset());
}
/**
* 处理异常消息
*/
private void handleProcessingException(ConsumerRecord<String, String> record, Throwable e) {
// 异常处理策略
logger.error("异常消息处理策略 - Topic: {}, Partition: {}, Offset: {}",
record.topic(), record.partition(), record.offset());
}
/**
* 停止消费者
*/
public void stop() {
running = false;
consumer.wakeup();
// 关闭线程池
executorService.shutdown();
try {
if (!executorService.awaitTermination(30, TimeUnit.SECONDS)) {
executorService.shutdownNow();
}
} catch (InterruptedException e) {
executorService.shutdownNow();
Thread.currentThread().interrupt();
}
logger.info("Kafka消费者停止完成");
}
/**
* 消息处理器接口
*/
public interface MessageProcessor {
ProcessingResult process(String key, String value, MessageMetadata metadata);
}
/**
* 处理结果
*/
public static class ProcessingResult {
private final boolean success;
private final String errorMessage;
public ProcessingResult(boolean success, String errorMessage) {
this.success = success;
this.errorMessage = errorMessage;
}
public static ProcessingResult success() {
return new ProcessingResult(true, null);
}
public static ProcessingResult error(String errorMessage) {
return new ProcessingResult(false, errorMessage);
}
public boolean isSuccess() { return success; }
public String getErrorMessage() { return errorMessage; }
}
/**
* 消息元数据
*/
public static class MessageMetadata {
private final String topic;
private final int partition;
private final long offset;
private final long timestamp;
// 构造函数和getter方法...
}
}
quote 💭 Java大数据技术栈实践心得
🎯 生态集成：充分利用Java在大数据生态中的优势和丰富的框架支持 💪 性能调优：重视JVM参数调优、内存管理和并发处理优化 🔍 监控运维：建立完善的监控体系，及时发现和解决生产环境问题 🛠️ 最佳实践：遵循企业级开发规范，注重代码质量和系统可维护性
相关学习资源
继续学习
- 学习路线图：Java
- Java在数据开发中的应用：Java在数据开发中的
- 数据库操作与数据仓库开发：Java数据库操作与数据仓库开发
大数据框架学习
- Hadoop生态系统：Hadoop/Hadoop
- Spark大数据处理：Spark/Spark
- Kafka消息中间件：Kafka/Kafka
- Flink流式计算：Flink/Flink
tip 🌟 技术栈建议
Java大数据技术栈的学习需要循序渐进，建议先掌握单机版本的使用，再深入集群部署和性能调优。重点关注数据处理的业务逻辑实现，同时注重系统的稳定性和可扩展性。在企业实践中，要特别关注数据质量、处理性能和运维监控。