mirror of https://github.com/halo-dev/halo
Refactor visit event
johnniang
parent
93eb1250f7
commit
4695868fa7
|
|
@ -0,0 +1,34 @@
|
|||
package run.halo.app.event.post;
|
||||
|
||||
import org.springframework.context.ApplicationEvent;
|
||||
import org.springframework.lang.NonNull;
|
||||
import org.springframework.util.Assert;
|
||||
|
||||
/**
|
||||
* Visit event.
|
||||
*
|
||||
* @author johnniang
|
||||
* @date 19-4-22
|
||||
*/
|
||||
public abstract class AbstractVisitEvent extends ApplicationEvent {
|
||||
|
||||
private final Integer id;
|
||||
|
||||
/**
|
||||
* Create a new ApplicationEvent.
|
||||
*
|
||||
* @param source the object on which the event initially occurred (never {@code null})
|
||||
* @param id id
|
||||
*/
|
||||
public AbstractVisitEvent(@NonNull Object source, @NonNull Integer id) {
|
||||
super(source);
|
||||
|
||||
Assert.notNull(id, "Id must not be null");
|
||||
this.id = id;
|
||||
}
|
||||
|
||||
@NonNull
|
||||
public Integer getId() {
|
||||
return id;
|
||||
}
|
||||
}
|
||||
|
|
@ -1,6 +1,12 @@
|
|||
package run.halo.app.event.post;
|
||||
|
||||
import cn.hutool.core.lang.Assert;
|
||||
import lombok.NonNull;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import run.halo.app.service.base.BasePostService;
|
||||
|
||||
import java.util.Map;
|
||||
import java.util.concurrent.*;
|
||||
|
||||
/**
|
||||
* Abstract visit event listener.
|
||||
|
|
@ -11,48 +17,102 @@ import lombok.extern.slf4j.Slf4j;
|
|||
@Slf4j
|
||||
public abstract class AbstractVisitEventListener {
|
||||
|
||||
// private final Map<Integer, BlockingQueue<Integer>> postVisitQueueMap;
|
||||
//
|
||||
// private final Map<Integer, PostVisitEventListener.PostVisitTask> postVisitTaskMap;
|
||||
//
|
||||
// protected final BasePostRepository basePostRepository;
|
||||
//
|
||||
// protected AbstractVisitEventListener(BasePostRepository basePostRepository) {
|
||||
// this.basePostRepository = basePostRepository;
|
||||
// }
|
||||
//
|
||||
//
|
||||
// /**
|
||||
// * Post visit task.
|
||||
// */
|
||||
// private class PostVisitTask implements Runnable {
|
||||
//
|
||||
// private final Integer postId;
|
||||
//
|
||||
// private PostVisitTask(Integer postId) {
|
||||
// this.postId = postId;
|
||||
// }
|
||||
//
|
||||
// @Override
|
||||
// public void run() {
|
||||
// while (!Thread.currentThread().isInterrupted()) {
|
||||
// try {
|
||||
// BlockingQueue<Integer> postVisitQueue = postVisitQueueMap.get(postId);
|
||||
// Integer postId = postVisitQueue.take();
|
||||
//
|
||||
// log.debug("Took a new visit for post id: [{}]", postId);
|
||||
//
|
||||
// // Increase the visit
|
||||
// postService.increaseVisit(postId);
|
||||
//
|
||||
// log.debug("Increased visits for post id: [{}]", postId);
|
||||
// } catch (InterruptedException e) {
|
||||
// log.debug("Post visit task: " + Thread.currentThread().getName() + " was interrupted", e);
|
||||
// // Ignore this exception
|
||||
// }
|
||||
// }
|
||||
//
|
||||
// log.debug("Thread: [{}] has been interrupted", Thread.currentThread().getName());
|
||||
// }
|
||||
// }
|
||||
private final Map<Integer, BlockingQueue<Integer>> postVisitQueueMap;
|
||||
|
||||
private final Map<Integer, PostVisitTask> postVisitTaskMap;
|
||||
|
||||
private final BasePostService basePostService;
|
||||
|
||||
private final ExecutorService executor;
|
||||
|
||||
protected AbstractVisitEventListener(BasePostService basePostService) {
|
||||
this.basePostService = basePostService;
|
||||
|
||||
int initCapacity = 8;
|
||||
|
||||
long count = basePostService.count();
|
||||
|
||||
if (count < initCapacity) {
|
||||
initCapacity = (int) count;
|
||||
}
|
||||
|
||||
postVisitQueueMap = new ConcurrentHashMap<>(initCapacity << 1);
|
||||
postVisitTaskMap = new ConcurrentHashMap<>(initCapacity << 1);
|
||||
|
||||
this.executor = Executors.newCachedThreadPool();
|
||||
}
|
||||
|
||||
/**
|
||||
* Handle visit event.
|
||||
*
|
||||
* @param event visit event must not be null
|
||||
* @throws InterruptedException
|
||||
*/
|
||||
protected void handleVisitEvent(@NonNull AbstractVisitEvent event) throws InterruptedException {
|
||||
Assert.notNull(event, "Visit event must not be null");
|
||||
|
||||
// Get post id
|
||||
Integer id = event.getId();
|
||||
|
||||
log.debug("Received a visit event, post id: [{}]", id);
|
||||
|
||||
// Get post visit queue
|
||||
BlockingQueue<Integer> postVisitQueue = postVisitQueueMap.computeIfAbsent(id, this::createEmptyQueue);
|
||||
|
||||
postVisitTaskMap.computeIfAbsent(id, this::createPostVisitTask);
|
||||
|
||||
// Put a visit for the post
|
||||
postVisitQueue.put(id);
|
||||
}
|
||||
|
||||
|
||||
private PostVisitTask createPostVisitTask(Integer postId) {
|
||||
// Create new post visit task
|
||||
PostVisitTask postVisitTask = new PostVisitTask(postId);
|
||||
// Start a post visit task
|
||||
executor.execute(postVisitTask);
|
||||
|
||||
log.debug("Created a new post visit task for post id: [{}]", postId);
|
||||
return postVisitTask;
|
||||
}
|
||||
|
||||
private BlockingQueue<Integer> createEmptyQueue(Integer postId) {
|
||||
// Create a new queue
|
||||
return new LinkedBlockingQueue<>();
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* Post visit task.
|
||||
*/
|
||||
private class PostVisitTask implements Runnable {
|
||||
|
||||
private final Integer postId;
|
||||
|
||||
private PostVisitTask(Integer postId) {
|
||||
this.postId = postId;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void run() {
|
||||
while (!Thread.currentThread().isInterrupted()) {
|
||||
try {
|
||||
BlockingQueue<Integer> postVisitQueue = postVisitQueueMap.get(postId);
|
||||
Integer postId = postVisitQueue.take();
|
||||
|
||||
log.debug("Took a new visit for post id: [{}]", postId);
|
||||
|
||||
// Increase the visit
|
||||
basePostService.increaseVisit(postId);
|
||||
|
||||
log.debug("Increased visits for post id: [{}]", postId);
|
||||
} catch (InterruptedException e) {
|
||||
log.debug("Post visit task: " + Thread.currentThread().getName() + " was interrupted", e);
|
||||
// Ignore this exception
|
||||
}
|
||||
}
|
||||
|
||||
log.debug("Thread: [{}] has been interrupted", Thread.currentThread().getName());
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,34 +1,20 @@
|
|||
package run.halo.app.event.post;
|
||||
|
||||
import org.springframework.context.ApplicationEvent;
|
||||
import org.springframework.lang.NonNull;
|
||||
import org.springframework.util.Assert;
|
||||
|
||||
/**
|
||||
* Visit event.
|
||||
* Post visit event.
|
||||
*
|
||||
* @author johnniang
|
||||
* @date 19-4-22
|
||||
*/
|
||||
public class PostVisitEvent extends ApplicationEvent {
|
||||
|
||||
private final Integer postId;
|
||||
public class PostVisitEvent extends AbstractVisitEvent {
|
||||
|
||||
/**
|
||||
* Create a new ApplicationEvent.
|
||||
*
|
||||
* @param source the object on which the event initially occurred (never {@code null})
|
||||
* @param postId post id
|
||||
* @param postId post id must not be null
|
||||
*/
|
||||
public PostVisitEvent(@NonNull Object source, @NonNull Integer postId) {
|
||||
super(source);
|
||||
|
||||
Assert.notNull(postId, "Post id must not be null");
|
||||
this.postId = postId;
|
||||
}
|
||||
|
||||
@NonNull
|
||||
public Integer getPostId() {
|
||||
return postId;
|
||||
public PostVisitEvent(Object source, Integer postId) {
|
||||
super(source, postId);
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -6,10 +6,6 @@ import org.springframework.scheduling.annotation.Async;
|
|||
import org.springframework.stereotype.Component;
|
||||
import run.halo.app.service.PostService;
|
||||
|
||||
import javax.annotation.PreDestroy;
|
||||
import java.util.Map;
|
||||
import java.util.concurrent.*;
|
||||
|
||||
/**
|
||||
* Visit event listener.
|
||||
*
|
||||
|
|
@ -18,103 +14,15 @@ import java.util.concurrent.*;
|
|||
*/
|
||||
@Slf4j
|
||||
@Component
|
||||
public class PostVisitEventListener {
|
||||
|
||||
private final Map<Integer, BlockingQueue<Integer>> postVisitQueueMap;
|
||||
|
||||
private final Map<Integer, PostVisitTask> postVisitTaskMap;
|
||||
|
||||
private final PostService postService;
|
||||
|
||||
private final ExecutorService executor;
|
||||
public class PostVisitEventListener extends AbstractVisitEventListener {
|
||||
|
||||
public PostVisitEventListener(PostService postService) {
|
||||
this.postService = postService;
|
||||
|
||||
int initCapacity = 8;
|
||||
|
||||
long count = postService.count();
|
||||
|
||||
if (count < initCapacity) {
|
||||
initCapacity = (int) count;
|
||||
}
|
||||
|
||||
postVisitQueueMap = new ConcurrentHashMap<>(initCapacity << 1);
|
||||
postVisitTaskMap = new ConcurrentHashMap<>(initCapacity << 1);
|
||||
|
||||
this.executor = Executors.newCachedThreadPool();
|
||||
super(postService);
|
||||
}
|
||||
|
||||
@Async
|
||||
@EventListener
|
||||
public void onApplicationEvent(PostVisitEvent event) throws InterruptedException {
|
||||
// Get post id
|
||||
Integer postId = event.getPostId();
|
||||
|
||||
log.debug("Received a visit event, post id: [{}]", postId);
|
||||
|
||||
// Get post visit queue
|
||||
BlockingQueue<Integer> postVisitQueue = postVisitQueueMap.computeIfAbsent(postId, this::createEmptyQueue);
|
||||
|
||||
postVisitTaskMap.computeIfAbsent(postId, this::createPostVisitTask);
|
||||
|
||||
// Put a visit for the post
|
||||
postVisitQueue.put(postId);
|
||||
|
||||
// TODO Attempt to manage the post visit tasks
|
||||
}
|
||||
|
||||
private PostVisitTask createPostVisitTask(Integer postId) {
|
||||
// Create new post visit task
|
||||
PostVisitTask postVisitTask = new PostVisitTask(postId);
|
||||
// Start a post visit task
|
||||
executor.execute(postVisitTask);
|
||||
|
||||
log.debug("Created a new post visit task for post id: [{}]", postId);
|
||||
return postVisitTask;
|
||||
}
|
||||
|
||||
@PreDestroy
|
||||
protected void preDestroy() {
|
||||
executor.shutdownNow();
|
||||
}
|
||||
|
||||
private BlockingQueue<Integer> createEmptyQueue(Integer postId) {
|
||||
// Create a new queue
|
||||
return new LinkedBlockingQueue<>();
|
||||
}
|
||||
|
||||
/**
|
||||
* Post visit task.
|
||||
*/
|
||||
private class PostVisitTask implements Runnable {
|
||||
|
||||
private final Integer postId;
|
||||
|
||||
private PostVisitTask(Integer postId) {
|
||||
this.postId = postId;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void run() {
|
||||
while (!Thread.currentThread().isInterrupted()) {
|
||||
try {
|
||||
BlockingQueue<Integer> postVisitQueue = postVisitQueueMap.get(postId);
|
||||
Integer postId = postVisitQueue.take();
|
||||
|
||||
log.debug("Took a new visit for post id: [{}]", postId);
|
||||
|
||||
// Increase the visit
|
||||
postService.increaseVisit(postId);
|
||||
|
||||
log.debug("Increased visits for post id: [{}]", postId);
|
||||
} catch (InterruptedException e) {
|
||||
log.debug("Post visit task: " + Thread.currentThread().getName() + " was interrupted", e);
|
||||
// Ignore this exception
|
||||
}
|
||||
}
|
||||
|
||||
log.debug("Thread: [{}] has been interrupted", Thread.currentThread().getName());
|
||||
}
|
||||
public void onPostVisitEvent(PostVisitEvent event) throws InterruptedException {
|
||||
handleVisitEvent(event);
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -0,0 +1,20 @@
|
|||
package run.halo.app.event.post;
|
||||
|
||||
/**
|
||||
* Sheet visit event.
|
||||
*
|
||||
* @author johnniang
|
||||
* @date 19-4-24
|
||||
*/
|
||||
public class SheetVisitEvent extends AbstractVisitEvent {
|
||||
|
||||
/**
|
||||
* Create a new ApplicationEvent.
|
||||
*
|
||||
* @param source the object on which the event initially occurred (never {@code null})
|
||||
* @param sheetId sheet id must not be null
|
||||
*/
|
||||
public SheetVisitEvent(Object source, Integer sheetId) {
|
||||
super(source, sheetId);
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,26 @@
|
|||
package run.halo.app.event.post;
|
||||
|
||||
import org.springframework.context.event.EventListener;
|
||||
import org.springframework.scheduling.annotation.Async;
|
||||
import run.halo.app.service.SheetService;
|
||||
|
||||
/**
|
||||
* Sheet visit event listener.
|
||||
*
|
||||
* @author johnniang
|
||||
* @date 19-4-24
|
||||
*/
|
||||
public class SheetVisitEventListener extends AbstractVisitEventListener {
|
||||
|
||||
protected SheetVisitEventListener(SheetService sheetService) {
|
||||
super(sheetService);
|
||||
}
|
||||
|
||||
@Async
|
||||
@EventListener
|
||||
|
||||
public void onSheetVisitEvent(SheetVisitEvent event) throws InterruptedException {
|
||||
handleVisitEvent(event);
|
||||
}
|
||||
|
||||
}
|
||||
|
|
@ -5,6 +5,7 @@ import org.springframework.data.domain.Page;
|
|||
import org.springframework.data.domain.Pageable;
|
||||
import org.springframework.stereotype.Service;
|
||||
import org.springframework.util.Assert;
|
||||
import run.halo.app.event.post.SheetVisitEvent;
|
||||
import run.halo.app.model.dto.post.SheetDetailDTO;
|
||||
import run.halo.app.model.dto.post.SheetListDTO;
|
||||
import run.halo.app.model.entity.Sheet;
|
||||
|
|
@ -72,8 +73,7 @@ public class SheetServiceImpl extends BasePostServiceImpl<Sheet> implements Shee
|
|||
|
||||
if (PostStatus.PUBLISHED.equals(status)) {
|
||||
// Log it
|
||||
// TODO Fatal bug here
|
||||
// eventPublisher.publishEvent(new PostVisitEvent(this, sheet.getId()));
|
||||
eventPublisher.publishEvent(new SheetVisitEvent(this, sheet.getId()));
|
||||
}
|
||||
|
||||
return sheet;
|
||||
|
|
|
|||
Loading…
Reference in New Issue