流计算中的状态管理是什么？请解释其作用和常用方法。

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class StateManagementExample {
    public static void main(String[] args) throws Exception {
        // 创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 创建数据流
        DataStream<Integer> stream = env.fromElements(1, 2, 3, 4, 5);

        // 定义状态描述符
        ValueStateDescriptor<Integer> stateDescriptor = new ValueStateDescriptor<>("sum", Integer.class);

        // 定义状态管理
        DataStream<Integer> result = stream.map(new StateManagementFunction(stateDescriptor));

        // 打印结果
        result.print();

        // 执行任务
        env.execute("State Management Example");
    }

    // 自定义函数，用于管理状态
    public static class StateManagementFunction implements MapFunction<Integer, Integer> {
        private final ValueStateDescriptor<Integer> stateDescriptor;

        public StateManagementFunction(ValueStateDescriptor<Integer> stateDescriptor) {
            this.stateDescriptor = stateDescriptor;
        }

        @Override
        public Integer map(Integer value) throws Exception {
            // 获取状态
            ValueState<Integer> state = getRuntimeContext().getState(stateDescriptor);

            // 更新状态
            Integer sum = state.value();
            if (sum == null) {
                sum = 0;
            }
            sum += value;
            state.update(sum);

            // 返回结果
            return sum;
        }
    }
}