Comments (16)
Thank you very much, I didn't expect you to complete it so well, again, this project is great!I can download the source code directly and use it!
from rqueue.
Hi Yuluo
Thanks for the request, In the Spring bases apps tasks can be scheduled using
Scheduled. Are you trying to do something else apart from running some job at some time or your use-case can't be handled using Scheduled? It would be nice if you can provide some example usecase.
from rqueue.
hi, sonus
Thank you for your question. In my use, I have some periodic emails,i need to dynamic increase this periodic emails to redis,Now I want to add a periodic task that needs to be decomposed into subtasks and added to the requeue through a time array.If I want to cancel all subtasks of this periodic task, I have to manually delete them one by one.Can you provide an api to meet the addition and cancellation of periodic tasks.
Many thanks!!
from rqueue.
Thanks for the explanation, sorry! I could not follow your example, what do you mean by
i need to dynamic increase this periodic emails to redis
Let's assume you're scheduling tasks at 30 seconds interval.
IIUC, you're scheduling N tasks at t, t+30, t+60, t+90 ...
for some reason if you want to cancel this task then you manually delete all enqueued tasks related to this task.
Do you always delete all related tasks or only some of them?
Please provide some example data with timestamp what're you doing and what're trying to achieve using Rqueue. Maybe there's an easy way to achieve the same.
from rqueue.
ok, i use rqueue 2.0.4 ,so I have no way to delete the tasks in redis. i think this way to realize cyclic tasks.
this is my webapp
public RouterFunction<ServerResponse> emailRouterquest(EmailHandleFuction handler){
RouterFunctions.Builder route = route();
route.POST("/email/messageByPeriodic", accept(MediaType.APPLICATION_JSON).and(contentType(MediaType.APPLICATION_JSON)), handler::asynPeriodic);
return route.build();
}
I use this api to accept periodic recurring tasks, it will accept a emailBean
public class EmailBean {
private String sendto;
// Used to identify tasks
private String id;
// Task start time
private String start;
// Period task end time
private String end;
// Time period
private String periodic;
}
In my processing function, I will add many tasks based on the start time, end time, and time interval
public Mono<ServerResponse> asynPeriodic(ServerRequest serverRequest) {
return serverRequest.bodyToMono(EmailBean.class).flatMap((Function<EmailBean, Mono<ServerResponse>>) emailBean -> {
if (emailBean.getEnd() != null && emailBean.getStart() != null && emailBean.getPeriodic() != null
&& !emailBean.getEnd().equals("") && !emailBean.getStart().equals("") && !emailBean.getPeriodic().equals("")) {
// Generate time list
ArrayList<Date> timeList = dateTime.getPeriodicTime(
emailBean.getStart(),
emailBean.getEnd(),
emailBean.getPeriodic()
);
for (Date time : timeList) {
log.warn("Add periodic tasks",time);
rqueueMessageSender.enqueueAt("delay-queue", emailBean, time);
}
}
return ServerResponse.ok()
.bodyValue(CommonResultImpl.success("Add periodic task"));
});
}
For example, a periodic task is Send email every 12 hours, from November 2020 to December 2020. so, it will add 124 task to rqueue. These tasks will have the same task id to indicate that they are the same cycle task.
@RqueueListener(
value ="${rqueue.delay.queue.name}",
numRetries = "5",
visibilityTimeout = "5*60*1000",
deadLetterQueue = "morgue-queue",
deadLetterQueueListenerEnabled = "true",
active = "true",
concurrency = "2-3"
)
public void onMessage(EmailBean emailBean){
/**
* When I want to cancel sending mail
* I will store the task id in the data table
* If id exists, no mail will be sent
*/
Mono<cancelList> info= cancelRepository.findTopByMessageId(emailBean.getId());
cancelList get = info.block();
if(get != null){
}else {
log.info("send email!!"+emailBean.toString());
emailUtil.send(emailBean);
}
}
When these task is executing, I sometimes need to cancel it. The task id of my assigned task will stored on the data table,When each subtask is executed, it will first query the database.If this id exists, the subtask will not be executed.
in this way, I implemented a cycle task, but This method is too bad.so Is it possible to add cycle task function.Or can you tell me how to solve similar problems.If you can tell me how to implement this feature, I can also help requeue implement it.
My English is not very good, please forgive me if I offend. This project is really great!!!
from rqueue.
in this way, I implemented a cycle task, but This method is too bad.
Why do you think so?
This is one of the easiest way to solve this problem. Other way could be you can use MessageProcessor (this can be set in SimpleRqueueListenerContainerFactory
via setPreExecutionMessageProcessor
) that should return false for tasks those can be ignored. Again here you can check id
of the EmailBean
to ignore such tasks.
from rqueue.
Because it will generate a lot of subtasks,and When I need to cancel, I need to rely on a third-party database.
My opinion is,rqueueMessageSender.enqueueBycycle("delay-queue", emailBean, timeInterval);
,When the task execution is completed, the task will re-enter zset,To achieve cyclic tasks.
If so, how can I achieve.
Thank you very much!
from rqueue.
Interesting, this can be achieved by setting isPeriodic
and interval
in RqueueMessage
, but this leaves us with some lingering questions related to retry and execution time.
For example
Task t
was enqueued at 7:30:00PM
with period 30
seconds
- If execution of task
t
at7:30:30PM
fails then it will be retried later at7:30:30+PM
=> this could potentially miss next task if it's scheduled at7:31:00PM
. - Success full execution of task
t
took45 seconds
than again it will miss the next task that should be scheduled at7:31:00PM
- If we enqueue message as soon the task is dequeued than this will create another problem of duplicate tasks due to retry e.g task was dequeued at
7:30:30PM
no matter what happens we'll enqueue a new task that should run at7:31:00PM
You can use similar approach in your listener code, like as soon as your listener is invoked you enqueue a new task at next interval. But what will you do when this message is retried, we can't identify exactly if a message is being retried, one way to identify this is whenever your listener is invoked you can check RqueueMessage#failureCount
but this is not 100% reliable.
from rqueue.
@yuluo-zy did this make sense or you would like to have such features in Rqueue?
from rqueue.
I am currently working on the source code,I want to achieve it,I initially considered whether to mark it as a cyclic task by adding attributes to RqueueMessage.
from rqueue.
Cool, let me know if you would like to see this feature in Rqueue, we would need to find some ways accurately find failure execution messages.
from rqueue.
I haven't finished reading the part of the code about the exception retry,but i think,If we enqueue message as soon the task is dequeued,By judging whether id is in zset.We can add without repeating.The retry after the task fails should not be related to the re-adding of the task.
from rqueue.
The problem here is we don;'t store id in zset instead we store entire RqueueMessage in zset, this makes it a difficult soln.
RqueueExecutor is the place to find relavant retry code.
from rqueue.
Hi
I have made changes to support periodic job via #50
Would you mind giving a shot, I can provide you a release candidate or snapshot.
from rqueue.
Please try and let me know, if something is breaking though I have tested using integration test.
from rqueue.
Thanks for the request, this feature has been added in 2.2.0 and it should be available in maven central. Open new bug for any issue etc.
from rqueue.
Related Issues (20)
- what is the sonatypeUsername and sonatypePassword? HOT 3
- Serialization support java time module HOT 1
- Deleted messages are kept in memory HOT 5
- Add static resource mapping handler when rqueue.web.url.prefix is set HOT 5
- Weighted priorisation with maximum concurrency HOT 2
- Question: Re-starting processing task when system restarts HOT 7
- Do you plan to use spring-cloud-sleuth to propagate the traceId? HOT 5
- Dashboard shows empty even thought Rqueue is working fine HOT 13
- Question: Integration scenario with multiple queues HOT 4
- rqueue workflow discussion HOT 4
- Support of jackson-databind 2.16.0 HOT 3
- Login/pass for queue dashboard HOT 4
- Lack of available threads HOT 2
- Service-Level Concurrency Control HOT 1
- IllegalArgumentException when accessing queue details HOT 1
- springboot 3.2 can not work HOT 5
- springboot 2.7 can not work HOT 9
- Typo in dashboard text HOT 2
- RQueue attempting to bootstrap despite being "disabled" HOT 1
- Best practice of initializing a queue (producer only) HOT 1
Recommend Projects
-
React
A declarative, efficient, and flexible JavaScript library for building user interfaces.
-
Vue.js
🖖 Vue.js is a progressive, incrementally-adoptable JavaScript framework for building UI on the web.
-
Typescript
TypeScript is a superset of JavaScript that compiles to clean JavaScript output.
-
TensorFlow
An Open Source Machine Learning Framework for Everyone
-
Django
The Web framework for perfectionists with deadlines.
-
Laravel
A PHP framework for web artisans
-
D3
Bring data to life with SVG, Canvas and HTML. 📊📈🎉
-
Recommend Topics
-
javascript
JavaScript (JS) is a lightweight interpreted programming language with first-class functions.
-
web
Some thing interesting about web. New door for the world.
-
server
A server is a program made to process requests and deliver data to clients.
-
Machine learning
Machine learning is a way of modeling and interpreting data that allows a piece of software to respond intelligently.
-
Visualization
Some thing interesting about visualization, use data art
-
Game
Some thing interesting about game, make everyone happy.
Recommend Org
-
Facebook
We are working to build community through open source technology. NB: members must have two-factor auth.
-
Microsoft
Open source projects and samples from Microsoft.
-
Google
Google ❤️ Open Source for everyone.
-
Alibaba
Alibaba Open Source for everyone
-
D3
Data-Driven Documents codes.
-
Tencent
China tencent open source team.
from rqueue.