这里写自定义目录标题
- 一、阻塞队列
- 简单的阻塞队列测试案例
- 总结
- 阻塞队列
- 二、Kafka入门
- 1、基础知识
- Kafka术语
- 消息队列实现方式两种
- 2、配置
- 3、启动
- 全部命令
- 启动 zookeeper 服务器
- 再启动 kafka 服务器
- 创建Topic
- 关闭
- 4、总结
- Kafka的特点
- Kafka的术语
- 三、 Spring整合Kafka
- 导入依赖
- application.properties
- KafkaTests.java
- 测试结果
- 四、发送系统通知
- 事件主体——Event
- 事件的生产者和消费者
- 补充评论mappper方法
- CommentMapper
- comment-mapper.xml
- CommentService
- 视图层评论——CommentController
- 视图层点赞——LikeController
- 视图层关注——FollowController
- 测试结果
- 五、显示系统通知
- 1、通知列表
- 数据访问层——MessageMapper.java
- 业务层——MessageService
- 视图层——MessageController
- 页面
- letter.html
- notice.html
- 测试
- 2、通知详情
- 数据访问层——MessageMapper.java
- 业务层——MessageService
- 视图层——MessageController
- 页面
- notice.html
- notice-detail.html
- 3、未读消息(总的)
- 拦截器——MessageInterceptor
- 拦截器配置——WebMvcConfig.java
- index.html
性能最好的消息队列
一、阻塞队列
阻塞队列——java自带,接口,BlockingQueue
二倍阻塞——消费者快
简单的阻塞队列测试案例
BlockingQueueTests
package com.nowcoder.community;
import java.util.Random;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
/**
* 阻塞队列测试案例
*/
public class BlockingQueueTests {
public static void main(String[] args) {
BlockingQueue queue = new ArrayBlockingQueue(10); // 队列容量
new Thread(new Producer(queue)).start();
new Thread(new Consumer(queue)).start();
new Thread(new Consumer(queue)).start();
new Thread(new Consumer(queue)).start();
}
}
/**
* 生产者线程
*/
// 实现接口
class Producer implements Runnable {
// 接收传来的阻塞队列
private BlockingQueue<Integer> queue;
public Producer(BlockingQueue<Integer> queue) {
this.queue = queue;
}
@Override
public void run() {
try {
for (int i = 0; i < 100; i++) { // 生产100数据
Thread.sleep(20); // 间隔时间 20ms
queue.put(i);
System.out.println(Thread.currentThread().getName() + "生产:" + queue.size());
}
} catch (Exception e) {
e.printStackTrace();
}
}
}
/**
* 消费者线程
*/
class Consumer implements Runnable {
private BlockingQueue<Integer> queue;
public Consumer(BlockingQueue<Integer> queue) {
this.queue = queue;
}
@Override
public void run() {
try {
while (true) {
Thread.sleep(new Random().nextInt(1000));
queue.take();
System.out.println(Thread.currentThread().getName() + "消费:" + queue.size());
}
} catch (Exception e) {
e.printStackTrace();
}
}
}
总结
阻塞队列
- 阻塞队列的接口为BlockingQueue,该接口有ArrayBlockingQueue、LinkedBlockingQueue等多个实现类。
- 阻塞队列包含put方法,用于向队列中存入数据,当队列已满时,该方法将阻塞
- 阻塞队列包含take方法,用于从队列中获取数据,当队列已空时,该方法将阻塞
二、Kafka入门
Kafka官网
1、基础知识
- 项目中只用到了- 消息系统的功能
- 将消息存在硬盘上,长久保存
- 硬盘空间大,比内存价格低
- 读取硬盘效率高低取决于对硬盘的使用
- 对硬盘顺序读写性能很高,高于对内存的随机读写
- 高可靠性——分布式服务器,集成部署
Kafka术语
Broker
:服务器Zookeeper
:独立软件应用,管理其他集群,Kafka有内置,也可以单独安Topic
:发布消息空间,存放消息Partition
:分区(看上图)Offset
:消息在分区内存在的索引Leader Replica
:主副本,数据备份,一个分区有多个副本Follower Replica
:从副本
消息队列实现方式两种
- 点对点:BlockingQueue
- 发布订阅模式:很多消费者同时订阅, Kafka
2、配置
3、启动
全部命令
# 启动zookeeper服务器:
C:\Users\dlmu>j:
J:\>cd J:\software\environment\kafka_2.11-2.3.0
# 启动服务器 (先启动zookeeper服务器,再启动kafka) !!!千万不要手动暴力关闭,用下面的命令关闭
J:\software\environment\kafka_2.11-2.3.0>bin\windows\zookeeper-server-start.bat config\zookeeper.properties
# 启动kafka服务器:
C:\Users\dlmu>j:
J:\>cd J:\software\environment\kafka_2.11-2.3.0
J:\software\environment\kafka_2.11-2.3.0>bin\windows\kafka-server-start.bat config\server.properties
# 创建主题
kafka-topics.bat --create --bootstrap-server localhost:9092 --replication-factor 1 --partitions 1 --topic test
# 查看当前服务器的主题
kafka-topics.bat --list --bootstrap-server localhost:9092
# 创建生产者,往指定主题上发消息
kafka-console-producer.bat --broker-list localhost:9092 --topic test
# 消费者
kafka-console-consumer.bat --bootstrap-server localhost:9092 --topic test --from-beginning
# 关闭zookeeper服务器
zookeeper-server-stop.bat
# 关闭kafka服务器
kafka-server-stop.bat
启动 zookeeper 服务器
C:\Users\dlmu>j:
J:\>cd J:\software\environment\kafka_2.11-2.3.0
J:\software\environment\kafka_2.11-2.3.0>bin\windows\zookeeper-server-start.bat config\zookeeper.properties
启动成功
然后不要关闭,再开一个cmd
再启动 kafka 服务器
C:\Users\dlmu>j:
J:\>cd J:\software\environment\kafka_2.11-2.3.0
J:\software\environment\kafka_2.11-2.3.0>bin\windows\kafka-server-start.bat config\server.properties
再新启动一个命令
创建Topic
生产者
在启动消费者
关闭
# 关闭zookeeper服务器
zookeeper-server-stop.bat
# 关闭kafka服务器
kafka-server-stop.bat
4、总结
Kafka的特点
- Kafka是一个分布式的流媒体平台。
- Kafka可以应用于消息系统、日志收集、用户行为追踪、流式处理等多种场景
- Kafka具有高吞吐量、消息持久化、高可靠性、高扩展性等优点
Kafka的术语
- Kafka集群中的每台服务器叫Broker,整个集群由Zookeeper进行管理
- Kafka采用发布订阅模式,每条消息都要发送到指定的Topic上
- 每个Topic可分为多个Partition,这样可以提高Kafka的并发执行能力
三、 Spring整合Kafka
- 满足生产者消费者模式
导入依赖
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka</artifactId>
</dependency>
application.properties
# KafkaProperties
spring.kafka.bootstrap-servers=localhost:9092 kafka 端口
spring.kafka.consumer.group-id=community-consumer-group 消费者组id
spring.kafka.consumer.enable-auto-commit=true 是否自动提交消费者的偏移量
spring.kafka.consumer.auto-commit-interval=3000 自动提交频率
消费者读消息按偏移量
KafkaTests.java
package com.nowcoder.community;
@RunWith(SpringRunner.class)
@SpringBootTest
@ContextConfiguration(classes = CommunityApplication.class)
public class KafkaTests {
@Autowired
private KafkaProducer kafkaProducer;
@Test
public void testKafka() {
kafkaProducer.sendMessage("test", "你好");
kafkaProducer.sendMessage("test", "在吗");
try {
Thread.sleep(1000 * 10);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
// spring容器管理
@Component
class KafkaProducer {
@Autowired
private KafkaTemplate kafkaTemplate;
/**
* 发送消息
* @param topic 消息主题
* @param content 消息内容
*/
public void sendMessage(String topic, String content) {
kafkaTemplate.send(topic, content);
}
}
@Component
class KafkaConsumer {
@KafkaListener(topics = {"test"})
public void handleMessage(ConsumerRecord record) {
System.out.println(record.value());
}
}
测试结果
四、发送系统通知
为什么用消息队列?
- 三类不同的事,可定义三类不同主题,事件发生后,将其包装为消息扔到响应队列中,当前线程就可以处理下一个请求,不用管后续业务,后续业务,由消费者处理
- 从技术角度来说,用的是
Kafka消息队列
,来解决问题 - 从业务角度来说,解决问题的方式是事件驱动,以事件为目标,为主题。
事件主体——Event
package com.nowcoder.community.entity;
import java.util.HashMap;
import java.util.Map;
/**
* 事件
*/
public class Event {
private String topic; // 主题
private int userId; // 事件的人
private int entityType; // 实体类型
private int entityId;
private int entityUserId; // 实体的作者(帖子)
// 处理其他数据时可能会有需要记录的
private Map<String, Object> data = new HashMap<>();
public String getTopic() {
return topic;
}
public Event setTopic(String topic) {
// 修改,返回此类事件,方便编写
this.topic = topic;
return this;
}
public int getUserId() {
return userId;
}
public Event setUserId(int userId) {
this.userId = userId;
return this;
}
public int getEntityType() {
return entityType;
}
public Event setEntityType(int entityType) {
this.entityType = entityType;
return this;
}
public int getEntityId() {
return entityId;
}
public Event setEntityId(int entityId) {
this.entityId = entityId;
return this;
}
public int getEntityUserId() {
return entityUserId;
}
public Event setEntityUserId(int entityUserId) {
this.entityUserId = entityUserId;
return this;
}
public Map<String, Object> getData() {
return data;
}
public Event setData(String key, Object value) {
// 分成 key ,value
this.data.put(key, value);
return this;
}
}
事件的生产者和消费者
消息表
红框部分内容是下图通知的内容
补充评论mappper方法
CommentMapper
/**
* 根据id查一个评论
* @param id
* @return
*/
Comment selectCommentById(int id);
comment-mapper.xml
<select id="selectCommentById" resultType="Comment">
select <include refid="selectFields"></include>
from comment
where id = #{id}
</select>
CommentService
public Comment findCommentById(int id) {
return commentMapper.selectCommentById(id);
}
视图层评论——CommentController
异步、并发
@RequestMapping(path = "/add/{discussPostId}", method = RequestMethod.POST)
public String addComment(@PathVariable("discussPostId") int discussPostId, Comment comment) {
comment.setUserId(hostHolder.getUser().getId()); // 当前用户id
comment.setStatus(0);
comment.setCreateTime(new Date()); // 当前时间
commentService.addComment(comment); // 添加
// 触发评论事件
Event event = new Event()
.setTopic(TOPIC_COMMENT)
.setUserId(hostHolder.getUser().getId())
.setEntityType(comment.getEntityType())
.setEntityId(comment.getEntityId())
.setData("postId", discussPostId);
if (comment.getEntityType() == ENTITY_TYPE_POST) {
DiscussPost target = discussPostService.findDiscussPostById(comment.getEntityId());
event.setEntityUserId(target.getUserId());
} else if (comment.getEntityType() == ENTITY_TYPE_COMMENT) {
// 查找评论
Comment target = commentService.findCommentById(comment.getEntityId());
event.setEntityUserId(target.getUserId());
}
eventProducer.fireEvent(event);
// 帖子详情页面 + 帖子id
return "redirect:/discuss/detail/" + discussPostId;
}
视图层点赞——LikeController
/**
* 点赞
* @param entityType:实体
* @param entityId:id
* @return
*/
@RequestMapping(path = "/like", method = RequestMethod.POST)
@ResponseBody
public String like(int entityType, int entityId, int entityUserId, int postId) {
User user = hostHolder.getUser(); // 当前用户
// 不登录无法访问——拦截器
// 点赞
likeService.like(user.getId(), entityType, entityId, entityUserId);
// 数量
long likeCount = likeService.findEntityLikeCount(entityType, entityId);
// 状态
int likeStatus = likeService.findEntityLikeStatus(user.getId(), entityType, entityId);
// 返回的结果 —— 给页面,map封装
Map<String, Object> map = new HashMap<>();
map.put("likeCount", likeCount);
map.put("likeStatus", likeStatus);
// 触发点赞事件
if (likeStatus == 1) {
Event event = new Event()
.setTopic(TOPIC_LIKE)
.setUserId(hostHolder.getUser().getId())
.setEntityType(entityType)
.setEntityId(entityId)
.setEntityUserId(entityUserId)
.setData("postId", postId);
eventProducer.fireEvent(event);
}
// 返回json格式数据
return CommunityUtil.getJSONString(0, null, map);
}
视图层关注——FollowController
/**
* 关注(异步)
* @param entityType
* @param entityId
* @return
*/
@RequestMapping(path = "/follow", method = RequestMethod.POST)
@ResponseBody
public String follow(int entityType, int entityId) {
User user = hostHolder.getUser();
followService.follow(user.getId(), entityType, entityId);
// 触发关注事件
Event event = new Event()
.setTopic(TOPIC_FOLLOW)
.setUserId(hostHolder.getUser().getId())
.setEntityType(entityType)
.setEntityId(entityId)
.setEntityUserId(entityId);
eventProducer.fireEvent(event);
// 异步请求
return CommunityUtil.getJSONString(0, "已关注!");
}
测试结果
五、显示系统通知
1、通知列表
显示评论、点赞、关注三种类型的通知
查询未读消息数量——controller
数据访问层——MessageMapper.java
// 查询某个主题下最新的通知
Message selectLatestNotice(int userId, String topic);
// 查询某个主题所包含的通知数量
int selectNoticeCount(int userId, String topic);
// 查询未读的通知的数量
int selectNoticeUnreadCount(int userId, String topic);
message-mapper.xml
<select id="selectLatestNotice" resultType="Message">
select <include refid="selectFields"></include>
from message
where id in (
select max(id) from message
where status != 2
and from_id = 1
and to_id = #{userId}
and conversation_id = #{topic}
)
</select>
<select id="selectNoticeCount" resultType="int">
select count(id) from message
where status != 2
and from_id = 1
and to_id = #{userId}
and conversation_id = #{topic}
</select>
<select id="selectNoticeUnreadCount" resultType="int">
select count(id) from message
where status = 0
and from_id = 1
and to_id = #{userId}
<if test="topic!=null">
and conversation_id = #{topic}
</if>
</select>
业务层——MessageService
public Message findLatestNotice(int userId, String topic) {
return messageMapper.selectLatestNotice(userId, topic);
}
public int findNoticeCount(int userId, String topic) {
return messageMapper.selectNoticeCount(userId, topic);
}
public int findNoticeUnreadCount(int userId, String topic) {
return messageMapper.selectNoticeUnreadCount(userId, topic);
}
视图层——MessageController
/**
* 查询通知
* @param model
* @return
*/
@RequestMapping(path = "/notice/list", method = RequestMethod.GET)
public String getNoticeList(Model model) {
User user = hostHolder.getUser();
// 查询评论类通知
Message message = messageService.findLatestNotice(user.getId(), TOPIC_COMMENT);
if (message != null) {
Map<String, Object> messageVO = new HashMap<>();
messageVO.put("message", message);
// 将JSON 对象还原为
String content = HtmlUtils.htmlUnescape(message.getContent());
Map<String, Object> data = JSONObject.parseObject(content, HashMap.class);
messageVO.put("user", userService.findUserById((Integer) data.get("userId")));
messageVO.put("entityType", data.get("entityType"));
messageVO.put("entityId", data.get("entityId"));
messageVO.put("postId", data.get("postId")); // 帖子id
int count = messageService.findNoticeCount(user.getId(), TOPIC_COMMENT);
messageVO.put("count", count);
int unread = messageService.findNoticeUnreadCount(user.getId(), TOPIC_COMMENT);
messageVO.put("unread", unread);
model.addAttribute("commentNotice", messageVO);
}
// 查询点赞类通知
message = messageService.findLatestNotice(user.getId(), TOPIC_LIKE);
if (message != null) {
Map<String, Object> messageVO = new HashMap<>();
messageVO.put("message", message);
String content = HtmlUtils.htmlUnescape(message.getContent());
Map<String, Object> data = JSONObject.parseObject(content, HashMap.class);
messageVO.put("user", userService.findUserById((Integer) data.get("userId")));
messageVO.put("entityType", data.get("entityType"));
messageVO.put("entityId", data.get("entityId"));
messageVO.put("postId", data.get("postId"));
int count = messageService.findNoticeCount(user.getId(), TOPIC_LIKE);
messageVO.put("count", count);
int unread = messageService.findNoticeUnreadCount(user.getId(), TOPIC_LIKE);
messageVO.put("unread", unread);
model.addAttribute("likeNotice", messageVO);
}
// 查询关注类通知
message = messageService.findLatestNotice(user.getId(), TOPIC_FOLLOW);
if (message != null) {
Map<String, Object> messageVO = new HashMap<>();
messageVO.put("message", message);
String content = HtmlUtils.htmlUnescape(message.getContent());
Map<String, Object> data = JSONObject.parseObject(content, HashMap.class);
messageVO.put("user", userService.findUserById((Integer) data.get("userId")));
messageVO.put("entityType", data.get("entityType"));
messageVO.put("entityId", data.get("entityId"));
int count = messageService.findNoticeCount(user.getId(), TOPIC_FOLLOW);
messageVO.put("count", count);
int unread = messageService.findNoticeUnreadCount(user.getId(), TOPIC_FOLLOW);
messageVO.put("unread", unread);
model.addAttribute("followNotice", messageVO);
}
// 查询未读消息数量(未读私信总数量,未读通知总数量)
int letterUnreadCount = messageService.findLetterUnreadCount(user.getId(), null);
model.addAttribute("letterUnreadCount", letterUnreadCount);
int noticeUnreadCount = messageService.findNoticeUnreadCount(user.getId(), null);
model.addAttribute("noticeUnreadCount", noticeUnreadCount);
return "/site/notice";
}
前面加上
页面
letter.html
notice.html
测试
2、通知详情
分页显示某一类主题所包含的通知
数据访问层——MessageMapper.java
/**
* 查询某个主题所包含的通知列表
* @param userId
* @param topic
* @param offset
* @param limit
* @return
*/
// 查询某个主题所包含的通知列表
List<Message> selectNotices(int userId, String topic, int offset, int limit);
<select id="selectNotices" resultType="Message">
select <include refid="selectFields"></include>
from message
where status != 2
and from_id = 1
and to_id = #{userId}
and conversation_id = #{topic}
order by create_time desc
limit #{offset}, #{limit}
</select>
业务层——MessageService
/**
* 通知列表
*/
public List<Message> findNotices(int userId, String topic, int offset, int limit) {
return messageMapper.selectNotices(userId, topic, offset, limit);
}
视图层——MessageController
@RequestMapping(path = "/notice/detail/{topic}", method = RequestMethod.GET)
public String getNoticeDetail(@PathVariable("topic") String topic, Page page, Model model) {
User user = hostHolder.getUser();
page.setLimit(5);
page.setPath("/notice/detail/" + topic);
page.setRows(messageService.findNoticeCount(user.getId(), topic));
List<Message> noticeList = messageService.findNotices(user.getId(), topic, page.getOffset(), page.getLimit());
List<Map<String, Object>> noticeVoList = new ArrayList<>();
if (noticeList != null) {
for (Message notice : noticeList) {
Map<String, Object> map = new HashMap<>();
// 通知
map.put("notice", notice);
// 内容
String content = HtmlUtils.htmlUnescape(notice.getContent());
Map<String, Object> data = JSONObject.parseObject(content, HashMap.class);
map.put("user", userService.findUserById((Integer) data.get("userId")));
map.put("entityType", data.get("entityType"));
map.put("entityId", data.get("entityId"));
map.put("postId", data.get("postId"));
// 通知作者
map.put("fromUser", userService.findUserById(notice.getFromId()));
noticeVoList.add(map);
}
}
model.addAttribute("notices", noticeVoList);
// 设置已读
List<Integer> ids = getLetterIds(noticeList);
if (!ids.isEmpty()) {
messageService.readMessage(ids);
}
return "/site/notice-detail";
}
页面
notice.html
notice-detail.html
3、未读消息(总的)
~在页面头部显示所有的未读消息数量
拦截器——MessageInterceptor
package com.nowcoder.community.controller.interceptor;
import com.nowcoder.community.entity.User;
import com.nowcoder.community.service.MessageService;
import com.nowcoder.community.util.HostHolder;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.web.servlet.HandlerInterceptor;
import org.springframework.web.servlet.ModelAndView;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
// 拦截器——总消息数量
@Component
public class MessageInterceptor implements HandlerInterceptor {
@Autowired
private HostHolder hostHolder;
@Autowired
private MessageService messageService;
@Override
public void postHandle(HttpServletRequest request, HttpServletResponse response, Object handler, ModelAndView modelAndView) throws Exception {
User user = hostHolder.getUser();
// 未读消息数量,两个
if (user != null && modelAndView != null) {
int letterUnreadCount = messageService.findLetterUnreadCount(user.getId(), null);
int noticeUnreadCount = messageService.findNoticeUnreadCount(user.getId(), null);
modelAndView.addObject("allUnreadCount", letterUnreadCount + noticeUnreadCount);
}
}
}
拦截器配置——WebMvcConfig.java
registry.addInterceptor(messageInterceptor)
.excludePathPatterns("/**/*.css", "/**/*.js", "/**/*.png", "/**/*.jpg", "/**/*.jpeg");