Schedule Endpoint is used to create and start scheduled tasks. Use cron expressions to define the task trigger time.

Type

endpoint/schedule

Startup configuration

None

Configure cron

router.From configures the cron expression. For example:

  //Execute every 1 second
  router1 := endpoint.NewRouter().From("*/1 * * * * *").Process(func(router *endpoint.Router, exchange *endpoint.Exchange) bool {
      exchange.In.GetMsg().Type = "TEST_MSG_TYPE1"
      fmt.Println("router1 执行...", time.Now().UnixMilli())
      //Business logic, such as reading files, regularly pulling some data to the rule chain for processing

      return true
  }).
//Specify which rule chain ID to process
To("chain:default").End()

Other parameters, refer to route Router definition.

cron format

cron expression represents a set of times, using 6 fields separated by spaces. For example: * * * * * * The fields from left to right represent:

Field	Required	Allowed values	Allowed special characters
Seconds	Yes	0-59	* / , -
Minutes	Yes	0-59	* / , -
Hours	Yes	0-23	* / , -
Day of month	Yes	1-31	* / , - ?
Month	Yes	1-12 or JAN-DEC	* / , -
Day of week	Yes	0-6 or SUN-SAT	* / , - ?

Note: The values of the Month and Day of week fields are not case sensitive. "SUN", "Sun" and "sun" are all allowed.

You can also use a few predefined variables to replace the cron expression.

Variable	Description	Equivalent to
@yearly (or @annually)	Run once at 0 o'clock on January 1 every year	0 0 0 1 1 *
@monthly	Run once at 0 o'clock on the 1st of every month	0 0 0 1 * *
@weekly	Run once at 0 o'clock every Sunday	0 0 0 * * 0
@daily (or @midnight)	Run once at 0 o'clock every day	0 0 0 * * *
@hourly	Run once every hour on the hour	0 0 * * * *

Example

scheduleEndpoint := endpoint.New(schedule.Type, config, nil)

//Execute every 1 second
router1 := endpoint.NewRouter().From("*/1 * * * * *").Process(func(router *endpoint.Router, exchange *endpoint.Exchange) bool {
  exchange.In.GetMsg().Type = "TEST_MSG_TYPE1"
  fmt.Println("router1 执行...", time.Now().UnixMilli())
  //Business logic, such as reading files, regularly pulling some data to the rule chain for processing
  return true
}).
//Specify which rule chain ID to process
To("chain:default").End()
//Add task route
routeId1, err := scheduleEndpoint.AddRouter(router1)
//Start task
err = scheduleEndpoint.Start()

The following are example codes using endpoint:

• RestEndpoint (opens new window)
• WebsocketEndpoint (opens new window)
• MqttEndpoint (opens new window)
• ScheduleEndpoint (opens new window)
• NetEndpoint (opens new window)
• KafkaEndpoint (opens new window) (Extended component library)

Multi-replica deployment

The cron runs inside each process, so every replica fires the same task in a multi-replica deployment. After injecting a distributed lock into the rule chain config (types.WithLocker), each scheduled slot is deduplicated across replicas and the cluster executes each planned time exactly once, without any DSL change. Clocks must be NTP-synchronized.

See Distributed Locker.