kafka第三次课！

package com.aaa.sbm.task;
import com.aaa.sbm.entity.Dept;
import com.aaa.sbm.service.DeptService;
import com.alibaba.fastjson.JSON;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.annotation.EnableAsync;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import org.springframework.web.client.RestTemplate;
import javax.annotation.Resource;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;
/**
• @ fileName:TimedSendDeptInfoTask
• @ description:
• @ author:zhz
• @ createTime:2022/1/13 9:37
• @ version:1.0.0/
@Component //不在3层之内，交给IOC处理
@EnableScheduling //开启定时任务
@EnableAsync //开启异步处理 可以在任务方法上使用@Async 该方法多线程处理时，可以异步处理，提高执行效率
@Slf4j
public class TimedSendDeptInfoTask {
//spring用到了哪些涉及模式 模板模式（封装出来一个通用的工具模板，供你完成什么功能，简化整个操作流程）
@Resource
private KafkaTemplate kafkaTemplate;
//juc包下线程安全的类，可以实现多线程同步自增
private AtomicInteger atomicInteger =new AtomicInteger();
//注入topic名称
@Value(“${topicName}”)
private String topicN;
/@Resourceprivate DeptService deptService;/
/@Resourceprivate RestTemplate restTemplate;/ //HttpClient
/@Resource //模板模式private RedisTemplate redisTemplate;*//**
• 定制执行任务
• 每隔3秒 使用多线程发送5条部门信息到kafka中
/
// cron=“秒 分 时 日 月 周” / 每 - 范围（3-10 每分钟的3秒到10秒每秒执行一次） , 选项 (3,10,15 每分钟第3秒第10秒第15秒执行)
@Scheduled(cron = "/3 * * * * ?")
public void timedExecute(){
//实例化固定线程池
ExecutorService executorService = Executors.newFixedThreadPool(5);
//lambda表达式 简化代码写法 ()->sendDeptInfo() 左边是参数 右边是执行业务
/* executorService.execute(new Runnable() {
@Override
public void run() {
sendDeptInfo();
System.out.println(“1”);
System.out.println(“2”);
}
});/
/ executorService.execute(()-> {
sendDeptInfo();
System.out.println(“1”);
System.out.println(“2”);
});*/
//启动5个线程执行
executorService.execute(()->sendDeptInfo());
executorService.execute(()->sendDeptInfo());
executorService.execute(()->sendDeptInfo());
executorService.execute(()->sendDeptInfo());
executorService.execute(()->sendDeptInfo());
//关闭线程池
executorService.shutdown();
}
/**
• 发送部门信息
*/
// @Async //异步处理，提高效率
public void sendDeptInfo(){
log.info(“线程信息为：”+Thread.currentThread().getName()+“,正在执行。。。。。。。。。。。。。。”);
int getAndIncrement = atomicInteger.getAndIncrement();
Dept dept =new Dept(getAndIncrement,“dev”+getAndIncrement,“zz”+getAndIncrement);
log.info(“要发送的部门信息为：”+dept);
//发送部门信息到kafka中 一定要是字符串格式
kafkaTemplate.send(topicN, JSON.toJSONString(dept));
}
}

package com.aaa.sbm.util;
import com.alibaba.fastjson.JSON;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
import java.util.Optional;
/**
• @ fileName:KafkaConsumer
• @ description:工具类用来监控topic-deptinof，不停的获取message
• @ author:zhz
• @ createTime:2022/1/13 10:26
• @ version:1.0.0*/@Component@Slf4jpublic class KafkaConsumer {/**
• 消费消息方法 借助 @KafkaListener指定消费的topic 如果该topic有信息都回被拉取pull 到参数中
• @param record
*/
@KafkaListener(topics = {“${topicName}”}) //监听注解 监听指定的topic
public void pullKafkaMsg(ConsumerRecord record){
//jdk8之后封装的专门处理空值一个类，有效防止空指针异常
Optional optional = Optional.ofNullable(record.value());
// isPresent等同于if(record!=null)
if(optional.isPresent()){
log.info(“接受到的信息为：”+ record);
log.info(“接受到的部门信息为：”+ optional.get());
}
}
}

@Autowired
private KafkaTemplate kafkaTemplate;
//发送消息方法
@GetMapping(“productOrder”)
public String send() {
Order order =new Order();
order.setId(100);
order.setMemberUsername(“测试生产者”);
order.setShopId(1001);
//log.info(“+++++++++++++++++++++ message = {}”, JSON.toJSONString(dept));
//topic-dept为主题
kafkaTemplate.send(“topic-order”, JSON.toJSONString(order));
return “suc”;
}

package com.aaa.ss.util;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
import java.util.Optional;
/**
• @ fileName:KafkaConsumer
• @ description:
• @ author:zhz
• @ createTime:2021/2/20 17:20
*/
@Component
@Slf4j
public class KafkaConsumer {
@KafkaListener(topics = {“topic-order”})
public void consumer(ConsumerRecord record){
Optional kafkaMessage = Optional.ofNullable(record.value());
if (kafkaMessage.isPresent()) {
Object message = kafkaMessage.get();
log.info(“----------------- record =” + record);
log.info(“------------------ message =” + message);
}
}
}