Flink中的Exactly-Once语义是什么？请解释其作用和实现原理。

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

public class ExactlyOnceExample {

    public static void main(String[] args) throws Exception {
        // 创建流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 设置事件时间
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        // 创建DataStream，从Kafka中接收用户访问数据流
        DataStream<UserVisitEvent> visitStream = env.addSource(new KafkaSource<>())
                .assignTimestampsAndWatermarks(new UserVisitEventTimestampExtractor());

        // 使用事件时间计算每个用户的访问次数
        DataStream<Tuple2<String, Long>> userCountStream = visitStream
                .keyBy(UserVisitEvent::getUser)
                .window(TumblingEventTimeWindows.of(Time.minutes(1)))
                .apply(new UserCountFunction());

        // 打印每个用户的访问次数
        userCountStream.print();

        // 执行流处理任务
        env.execute("Exactly-Once Example");
    }
}

class UserVisitEvent {
    private String user;
    private String page;
    private long timestamp;

    // 省略构造函数、getter和setter
}

class UserVisitEventTimestampExtractor extends BoundedOutOfOrdernessTimestampExtractor<UserVisitEvent> {
    public UserVisitEventTimestampExtractor() {
        super(Time.seconds(10)); // 设置最大延迟时间为10秒
    }

    @Override
    public long extractTimestamp(UserVisitEvent event) {
        // 返回事件的时间戳
        return event.getTimestamp();
    }
}

class UserCountFunction implements WindowFunction<UserVisitEvent, Tuple2<String, Long>, String, TimeWindow> {
    @Override
    public void apply(String user, TimeWindow window, Iterable<UserVisitEvent> events, Collector<Tuple2<String, Long>> out) {
        // 检查用户是否已经计算过访问次数
        boolean counted = checkUserCounted(user);

        // 如果用户还未计算过访问次数，则进行计算
        if (!counted) {
            // 计算用户的访问次数
            long count = 0;
            for (UserVisitEvent event : events) {
                count++;
            }

            // 输出结果
            out.collect(new Tuple2<>(user, count));

            // 更新用户计算状态为已计算
            updateUserCounted(user);
        }
    }

    private boolean checkUserCounted(String user) {
        // 查询用户计算状态，判断是否已经计算过访问次数
        // 返回true表示已经计算过，返回false表示还未计算过
    }

    private void updateUserCounted(String user) {
        // 更新用户计算状态为已计算
    }
}