如何保证我们的定时任务只会触发一次
Posted
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了如何保证我们的定时任务只会触发一次相关的知识，希望对你有一定的参考价值。
参考技术A 上一篇 <<< 传统任务调度存在的缺陷下一篇 >>> 定时任务宕机后如何处理
推荐阅读：
<<< 传统任务调度实现方案及demo示例
<<< 传统任务调度存在的缺陷
<<< 定时任务宕机后如何处理
<<< 分布式任务调度平台的特点
<<< XXL-Job和Elastic-job的区别
<<< Elastic-job的设计理念及原理说明
<<< SpringBoot整合Elastic-Job示例
<<< 修改Elastic-Job默认session过期时间提升故障转移速度
<<< XXL-job入门介绍
<<< XXL-job的架构与执行逻辑说明
<<< XXL-Job Admin平台搭建
<<< SpringBoot项目集成XXL-Job
<<< XXL-Job Admin任务调度的配置及预警示例
<<< XXL-Job使用分片集群机制将100W的短信在5min内发完
<<< XXL-Job如何搭建集群
SIGALRM信号设置定时器
alarm和setitimer函数设置的实时闹钟，一旦超时，将触发SIGALRM信号（只触发一次）。因此我们可以利用该信号的信号处理函数来处理定时任务。但是如果要处理多个定时任务，我们就需要不断的触发SIGALRM信号，并在其信号处理函数中执行到期的任务。
一般而言，SIGALRM信号按照固定的频率生成，即由alarm或setitmer函数设置的定时周期T保持不变。
这里我们将通过一个实例——处理非活动链接，来介绍SIGALRM信号定时。
并且使用一种简单的定时器实现——基于升序链表的定时器
lst_timer.h：
#ifndef LST_TIMER
#define LST_TIMER
#include <time.h>
#include <stdio.h>
#include <arpa/inet.h>
#define BUFFER_SIZE 64
class util_timer;
//用户数据结构：客户端socket地址、socket文件描述符、读缓存和定时器
struct client_data {
sockaddr_in address;
int sockfd;
char buf[BUFFER_SIZE];
util_timer* timer;
};
// 定时器类
class util_timer {
public:
util_timer() : prev(NULL), next(NULL) {}
public:
time_t expire; // 任务的超时时间，这里使用的是绝对时间
void (*cb_func) (client_data*); // 任务回调函数
// 回调函数处理的客户数据，由定时器的执行者传递给回调函数
util_timer* prev; // 指向前一个定时器
util_timer* next; // 指向下一个定时器
client_data* user_data;
};
// 定时器链表。他是一个升序、双向链表，且带有头结点和尾节点
class sort_timer_lst{
public:
sort_timer_lst() : head(NULL), tail(NULL) {}
// 链表销毁时，删除其中所有的定时器
~sort_timer_lst(){
util_timer* tmp = head;
while (tmp) {
head = tmp->next;
delete tmp;
tmp = head;
}
}
// 将目标定时器timer添加到链表中
void add_timer(util_timer* timer) {
if (!timer) return;
if (!head) {
head = tail = timer;
return;
}
/*
如果目标定时器的超时时间小于当前链表中所有定时器的超时时间，
则把该定时器插入链表头部作为新链表的头结点。
否则就需要调用重载函数add_timer(util_timer* timer, util_timer* lst_head)
把它插入链表中合格的位置，以确保链表的升序特性
*/
if (timer->expire < head->expire) {
timer->next = head;
head->prev = timer;
head = timer;
return;
}
add_timer(timer, head);
}
/*
当某个定时任务发生变化时，调整对应的定时器在链表中的位置
这个函数只考虑被调整的定时器的超市时间延长的情况
即该定时器需要往链表的尾部移动
*/
void adjust_timer(util_timer* timer) {
if (!timer) return;
util_timer* tmp = timer->next;
// 如果被调整的目标定时器处在链表尾部，或者改定时器洗呢超时值仍然小于其下一个定时器的超时值，则不用调整
if (!tmp || (timer->expire < tmp->expire)) return ;
// 如果目标定时器是链表的头结点，则将改定时器从链表中取出并重新插入链表
if (timer == head) {
head = head->next;
head->prev = NULL;
timer->next = NULL;
add_timer(timer, head);
}
// 如果目标定时器不是链表的头结点，则将改定时器从链表中取出，然后插入其原来所在位置之后的部分链表中
else {
timer->prev->next = timer->next;
timer->next->prev = timer->prev;
add_timer(timer, timer->next);
}
}
// 将目标定时器timer从链表中删除
void del_timer(util_timer* timer) {
if (!timer) return;
// 下面这个条件成立表示链表中只有一个定时器，即目标定时器
if ((timer == head) && (timer == tail)) {
delete timer;
head = NULL;
tail == NULL;
return;
}
// 如果链表中至少有两个定时器，且目标定时器是链表的头结点，
// 则将链表的头结点重置为原头结点的下一个节点，然后删除目标定时器
if (timer == head) {
head = head->next;
head->prev = NULL;
delete timer;
return;
}
// 如果链表中至少有两个定时器，且目标定时器是链表的尾节点
// 则将链表的尾节点重置为原尾节点的前一个节点，然后删除目标定时器
if (timer == tail) {
tail = tail->prev;
tail->next = NULL;
delete timer;
return;
}
// 如果目标定时器位于链表的中间，
// 则把它前后的定时器串联起来，然后删除目标定时器
timer->prev->next = timer->next;
timer->next->prev = timer->prev;
delete timer;
}
// SIGALRM信号每次被处罚就在其信号处理函数（如果使用统一事件源，则是主函数）中执行一次tick函数
// 以处理立案表上到期的任务
void tick() {
if (!head) return;
printf("timer tick\\n");
time_t cur = time(NULL); // 获取系统当前时间
util_timer* tmp = head;
// 从头节点开始一次处理每个定时器，直到遇到一个尚未到期的定时器
// 这就是定时器的核心逻辑
while (tmp) {
// 因为每个定时器都是用绝对时间作为超时值，
// 所以我们可以把定时器的超时值和系统当前时间，比较以判断定时器是否到期
if (cur < tmp->expire) break;
// 调用定时器的回调函数，以执行定时任务
tmp->cb_func(tmp->user_data);
// 执行完定时器中的定时任务之后，就将它从链表中删除，并重置头结点
head = tmp->next;
if (head) {
head->prev = NULL;
}
delete tmp;
tmp = head;
}
}
private:
// 一个重载的辅助函数，它的公有的add_timer函数和adjust_timer函数调用
// 还函数表示将目标定时器timer添加到节点lst_head之后的部分链表中
void add_timer(util_timer* timer, util_timer* lst_head) {
util_timer* prev = lst_head;
util_timer* tmp = prev->next;
// 遍历lst_head节点之后的部分链表，
// 直到找到一个超时时间大于目标定时器的超时时间的节点，并将目标定时器插入该节点之前
while(tmp) {
if (timer->expire < tmp->expire) {
prev->next = timer;
timer->next = tmp;
tmp->prev = timer;
timer->prev = prev;
break;
}
prev = tmp;
tmp = tmp->next;
}
// 如果便利玩lst_head节点之后的部分链表，
// 仍未找到超时时间大于目标定时器的超时时间的节点
// 则将目标定时器插入链表尾部，并把它设置为链表新的尾节点
if (!tmp) {
prev->next = timer;
timer->prev = prev;
timer->next = NULL;
tail = timer;
}
}
util_timer* head;
util_timer* tail;
};
#endif
signal_timer_test.cpp：
#include <sys/socket.h>
#include <sys/types.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <assert.h>
#include <stdio.h>
#include <unistd.h>
#include <errno.h>
#include <string.h>
#include <fcntl.h>
#include <stdlib.h>
#include <poll.h>
#include <libgen.h>
#include "lst_timer.h"
#include <sys/epoll.h>
#include <signal.h>
#define FD_LIMIT 65535
#define MAX_EVENT_NUMBER 1024
#define TIMESLOT 5
#define BACKLOG 5
static int pipefd[2];
// 利用lst_timer.h中的升序链表来管理定时器
static sort_timer_lst timer_lst;
static int epollfd = 0;
int setnonblocking (int fd) {
int old_option = fcntl(fd, F_GETFL);
int new_option = old_option | O_NONBLOCK;
fcntl(fd, F_SETFL, new_option);
return old_option;
}
void addfd (int epollfd, int fd) {
epoll_event event;
event.data.fd = fd;
event.events = EPOLLIN | EPOLLET;
epoll_ctl(epollfd, EPOLL_CTL_ADD, fd, &event);
setnonblocking(fd);
}
void sig_handler(int sig) {
int save_errno = errno;
int msg = sig;
send(pipefd[1], (char*)&msg, 1, 0);
errno = save_errno;
}
void addsig(int sig) {
struct sigaction sa;
memset(&sa, '\\0', sizeof(sa));
sa.sa_handler = sig_handler;
sa.sa_flags |= SA_RESTART;
sigfillset(&sa.sa_mask);
assert(sigaction(sig, &sa, NULL) != -1);
}
void timer_handler() {
// 定时处理任务，实际上就是调用tick函数
timer_lst.tick();
// 因为一次alarm调用只会引起一次SIGALRM信号，所以我们要重新定时，以不断触发SIGALRM信号
alarm(TIMESLOT);
}
// 定时器回调函数，它删除非活动连接socket上的注册事件，并关闭
void cb_func(client_data* user_data) {
epoll_ctl(epollfd, EPOLL_CTL_DEL, user_data->sockfd, 0);
assert(user_data);
close(user_data->sockfd);
printf("close fd %d\\n", user_data->sockfd);
}
int main(int argc, char* argv[]) {
if (argc <= 2) {
printf("usage: %s ip_address port_number\\n", basename(argv[0]));
return 1;
}
const char* ip = argv[1];
int port = atoi(argv[2]);
struct sockaddr_in address;
bzero(&address, sizeof(address));
address.sin_family = AF_INET;
inet_pton(AF_INET, ip, &address.sin_addr);
address.sin_port = htons(port);
int listenfd = socket(PF_INET, SOCK_STREAM, 0);
assert(listenfd >= 0);
int ret = bind(listenfd, (struct sockaddr*)&address, sizeof(address));
assert(ret != -1);
ret = listen(listenfd, BACKLOG);
assert(ret != -1);
epoll_event events[MAX_EVENT_NUMBER];
int epollfd = epoll_create(1);
assert(epollfd != -1);
addfd(epollfd, listenfd);
ret = socketpair(PF_UNIX, SOCK_STREAM, 0, pipefd); // 创建管道
assert(ret != -1);
setnonblocking(pipefd[1]);
addfd(epollfd, pipefd[0]);
//设置信号处理函数
addsig(SIGALRM);
addsig(SIGTERM);
bool stop_server = false;
client_data* users = new client_data[FD_LIMIT];
bool timeout = false;
alarm(TIMESLOT); // 设置定时周期，即TIMESLOT为一个周期。一个周期触发一次tick()函数
while (!stop_server) {
int number = epoll_wait(epollfd, events, MAX_EVENT_NUMBER, -1);
if ((number < 0) && (errno != EINTR)) {
printf("epoll failure");
break;
}
for (int i = 0; i < number; i++) {
int sockfd = events[i].data.fd;
// 处理新到的客户连接
if (sockfd == listenfd) {
struct sockaddr_in client_address;
socklen_t client_addresslen = sizeof(client_address);
int connfd = accept(sockfd, (struct sockaddr*)&client_address, &client_addresslen);
addfd(epollfd, connfd);
users[connfd].address = client_address;
users[connfd].sockfd = connfd;
// 创建定时器，设置其回调函数与超时时间，
// 然后绑定定时器与用户数据，最后将定时器添加到链表timer_lst中
util_timer* timer = new util_timer;
timer->user_data = &users[connfd];
timer->cb_func = cb_func;
time_t cur = time(NULL);
timer->expire = cur + 3 * TIMESLOT;
users[connfd].timer = timer;
}
// 处理信号
else if ((sockfd == pipefd[0]) && (events[i].events & EPOLLIN)) {
int sig;
char signals[1024];
ret = recv(pipefd[0], signals, sizeof(signals), 0);
if (ret == -1) {
// handle the error
continue;
}
else if (ret == 0) {
continue;
}
else {
for (int i = 0; i < ret; i++) {
switch (signals[i])
{
case SIGALRM:
{
// 用timeout变量标记有定时任务需要处理，但不立即处理定时任务。
// 这是因为定时任务的 优先级不是很高，我们优先处理其他更重要的任务
timeout = true;
break;
}
case SIGTERM:
{
stop_server = true;
}
}
}
}
}
// 处理客户连接上接收到的数据
else if (events[i].events & EPOLLIN) {
memset(users[sockfd].buf, '\\0', BUFFER_SIZE);
ret = recv(sockfd, users[sockfd].buf, BUFFER_SIZE - 1, 0);
printf("get %d bytes of client data %s from %d\\n", ret, users[sockfd].buf, sockfd);
util_timer* timer = users[sockfd].timer;
if (ret < 0) {
// 如果发生读错误，则关闭连接，并移除其对应的定时器
if (errno != EAGAIN) {
cb_func(&users[sockfd]);
if (timer) {
timer_lst.del_timer(timer);
}
}
}
else if (ret &#以上是关于如何保证我们的定时任务只会触发一次的主要内容，如果未能解决你的问题，请参考以下文章