【Godot4.5】竖版无限地图地形生成 流式地图生成性优化 Part-2

---

前言

我们在Part-1实现了噪声->地图的流程，并能实现基本的tilemap的填充以及通过创建子线程的方式把复杂计算下放给子线程处理来优化性能。经测试，在区块加载的时候一次性加载全部的tiles会造成帧间隔变大而造成掉帧阻塞主线程造成玩家体感下降的问题，所以有必要对加载tiles的流程进行优化，尤其对于要求Frame Per seconds要求较高的地图生成的场景。

---

场景说明

对于大部分由大量地图，大量元素构成的开放世界/rouguelite等游戏，围绕玩家为中心，某些场景内有大量的物品/物体/地形，如果立即加载会造成瞬时高发的内存占用，不可避免的造成帧渲染的性能降低，影响用户游戏体感。

为此本章提出的一种类Streaming技术/分片或批次加载在很多方面有广泛的应用。chunk加载实际上也是这个的一种（分片），对于简单/小型地图来讲chunk分的越细，加载chunk的时候造成的性能消耗可以忽略，对于像Godot的TileMap实现批量填充的时候，会占用大量的性能，因此需要再进行细分，甚至做到动态分片或分批，达到流式处理的效果。

原理

既然在一帧内渲染全部的tiles会造成线程阻塞，而TileMapLayer.set_cell等方法又要求必须在主线程中运行（前文已经提过，这里不再阐述原因。）
我们可以渲染tiles下方到每一帧去处理，每一帧仅处理一小部分。

生成优化

我们依然是以Chunk为主，把chunk再进行拆分处理，可以把chunk再拆一层，当该片所有的tile被填充了之后，标记该chunk被处理完毕。

var pending_tiles: Array[Dictionary] = []

定义一个全局变量用来存储挂起chunk当中需要填充的所有地形数据 terrain和chunk的 index

改造generate_chunk方法，把apply_chunk方法融合简化:

func generate_chunk(chunk_ind: int):
	var h_noise = FastNoiseLite.new()
	h_noise.seed = world_seed + 2026
	h_noise.noise_type = FastNoiseLite.TYPE_PERLIN
	h_noise.frequency = 0.1
	
	var t_noise = FastNoiseLite.new()
	t_noise.seed = world_seed - 2026
	t_noise.noise_type = FastNoiseLite.TYPE_PERLIN
	t_noise.frequency = 0.05
	
	var tile_data: Dictionary = {}
	for x in range(CHUNK_SIZE.x):
		for y in range(CHUNK_SIZE.y):
			var world_pos = Vector2i(get_chunk_start_x() + x, get_chunk_start_y(chunk_ind) + y)
			
			var cx = world_pos.x * 0.5
			var cy = world_pos.y * 0.5
			var h = h_noise.get_noise_2d(cx, cy)
			var t = t_noise.get_noise_2d(cx, cy)
			tile_data[world_pos] = calc_terrain_type(h,t)
	var terrain: Dictionary[int, Array] = { 
		0: [],
		1: [],
		2: []
	}
	
	for pos in tile_data:
		if terrain_tilemap.get_cell_source_id(pos) != -1:
			continue
		match tile_data[pos]:
			0: terrain[0].append(pos)  # water
			1, 2: terrain[1].append(pos)  # land
			3, 4: terrain[2].append(pos)  # mud
	if loaded_chunks.has(chunk_ind):
		return
	pending_tiles.append({
		index = chunk_ind,
		terrain = terrain
	})

我们改用dict<int, Array<vector2i>> 来存储原来的tiles_data

实现流式效果（无体感）

在一帧渲染了之后，我们需要决定一帧改处理多少个tiles，为了确保帧数，需要确保一个最大的处理时间。

关联 delta

假设 我们预期60帧/秒，及60FPS，一个chunk为15 * 45 = 675 个tiles
我们计划一秒内完成地图渲染。

这里为什么用1秒渲染完是按照视口高度和卷轴/滚轮移动速度计算出来的
要想确保视觉上的地图连贯，生成地图的速度要大于移动速度才行，即生产速度要大于消费速度
即 chunk_vh * chunk_gen_per_seconds > scroll_speed

每帧我们需要处理每帧填充11.25个tile才能满足1秒内渲染完地图。

我们用batch_size表示一帧内所需要填充的tiles的批次大小

消费因子

godot在process函数提供了一个delta参数，指的是上一帧到当前帧的运行时间，可以大致代表上一次渲染后_process运行时间，这个值通常以当前设备的渲染性能以及帧数上线相关，性能较好的设备，该值一般比较低，FPS可以达到60或更高。这个参数其实就是构成消费速度的因子。

因此我们需要在帧数比较低的时候即delta比较大的情况下，尽量增加渲染的批次数量（确保生产速度要大于消费速度），来确保帧数不会因为处理而造成损失，这不可避免的造成batch_size的下降来弥补损失。

计算每帧渲染的batch_size（生产速度）

为了确保不同性能的设备/屏幕的选滚轴滚动速度保持一致，实现每秒移动固定的路程，在主函数中，我们设置。

func _process(delta: float) -> void:
	$World.scroll_progress += delta * scroll_speed

确保world节点的scroll_progress每秒变化一致。
此时scroll_speed * delta表示一帧我们可以走多少像素点(pixel point)
我们把该值作为消费速度。

请注意，因为scroll_speed是与delta已经关联的了，可以根据帧数变化动态调整了，所以下面计算批次大小的时候，我们没必要再对此进行额外的一次运算了。

我们需要计算滚轴走过一个chunk需要多少帧，便于在滚轴移动的过程中，同步地形生成的速度(每帧或每批次填充的tiles数量，也就是基础的batch_size)。

我们用变量frame_per_chunk，滚轴走过一个chunk需要多少帧

var frame_per_chunk = CHUNK_SIZE.y * TILE_SIZE_PIX/ (game_instant.scroll_speed * delta)

其中TILE_SIZE_PIX为每个tile的大小（像素点）

我们期望在滚轴走完一个chunk的这些帧数之内，填充完地形。
我们需要填充的地形tiles数量为

var chunk_size = CHUNK_SIZE.x * CHUNK_SIZE.y

所以计算而来的batch_size = chunk_size / frame_per_chunk

或者简化成batch_size = CHUNK_SIZE.x * game_instant.scroll_speed * delta / TILE_SIZE_PIX

批次大小的微量增加不会太多的影响delta或者说渲染效率，所以我们向上取整同时额外+1作为批次渲染的容错，构成最终的batch_size = ceil(chunk_size / frame_per_chunk) + 1

批处理函数

我们期望每个帧间隔期间，分配一定的资源来进行tiles填充。
我们用max_time_out_usec表示为了维持某一帧所允许tiles填充行为的最大允许的允许时间(微妙)，以维持FPS

我们允许在某一帧进行1次或多次的tiles的批次填充，超过一次的填充行为是建立在上一批次batch_size没有用完的情况下的。
我们用batch_left表示进行一轮后剩余的可以用于填充的tiles数量

定义t0记录进行每帧填充的开始时间。
下方是本函数域变量的定义和初始化。

var t0 = Time.get_ticks_usec()
var batch_left = batch_size

我们需要持续扫描pending_tiles数组确保当之前线程执行generate_chunk方法生成了所需填充地形的tiles能被持续处理直到该列表为空。同时对该数组进行解包

while pending_tiles.size():
		var chunk = pending_tiles[0]
		var idx = chunk.index
		var terrain = chunk.terrain as Dictionary[int, Array]
		#...
t0 = Time.get_ticks_usec()

下面对每一批进行处理。
我们以一个地形类型所需填充的所有tiles作为被分片/拆分的父块。
我们以consume作为当前批的消耗，表示当前批填充处理的tiles的vector2i信息的大小
sliced作为当前批填充处理的tiles的vector2i信息的拆分单元
同时确保当前批不会超出处理（或者说越界保护）

var consume = min(batch_left, arr.size())
var sliced = arr.slice(0, consume)

对地形进行填充

terrain_tilemap.set_cells_terrain_connect(sliced, 0, type_id, 0)

在当前一次批处理完毕之后，我们需要裁剪掉之前填充的tiles，标记为已完成，保留未完成的部分。
同时计算当前批处理的剩余批可供下一批处理的数量以及当前批处理的时间消耗后的剩余允许的budget_left。

for type_id in terrain.keys():
			var arr: Array = terrain[type_id] as Array[Vector2i]
			if arr.is_empty(): continue
			var t1 = Time.get_ticks_usec()
			var consume = min(batch_left, arr.size())
			var sliced = arr.slice(0, consume)
			terrain_tilemap.set_cells_terrain_connect(sliced, 0, type_id, 0)
			
			var remain = arr.slice(consume, arr.size())	
			arr.clear()
			arr.append_array(remain)
			batch_left -= consume
			
			var budget_left = max_time_usec_out - (Time.get_ticks_usec() - t0)

如果发现当前批的数量已经填充完毕，或者当前批处理的时间消耗后的剩余允许的budget_left不足，那么只能交给下一帧进行处理。

if batch_left <= 0: continue
if budget_left <= 0: return

如果当前批次处理的数量还有剩余，那么就说明处理的当前chunk的所有类型的tiles被填充完毕。
我们清除挂机处理的chunk。

# 这一块是属于 while pending_tiles.size():下
if batch_left >= 0:
			pending_tiles.pop_at(0)
			loaded_chunks[idx] = true
			pending_chunks.erase(idx)
			# clean thread
			for i in range(active_threads.size()):
				var th = active_threads[i]
				if not th.is_started():
					th.wait_to_finish()
					active_threads.remove_at(i)
					break

完整代码

extends Node2D
@onready var terrain_tilemap: TileMapLayer = $TerrainTileMapLayer
@onready var stuff_tilemap: TileMapLayer = $StuffTileMapLayer
@onready var game_instant: Node = $".."
@export var LAND_FACTOR = 0.3
@export var scroll_progress: float = 0.0
@export var world_seed = 2026210

const CHUNK_SIZE = Vector2i(15, 40);
const PRELOAD_CHUNKS = 3
const TILE_SIZE_PIX = 16

var loaded_chunks = {}
var debug_last_index: int

var active_threads: Array[Thread] = []  
var pending_chunks: Array[int] = []

var pending_tiles: Array[Dictionary] = []

const TILE_TYLE = {
	0: 'boundry',
	1: 'inland',
}


func _ready() -> void:
	terrain_tilemap.set_cells_terrain_connect([Vector2i(0, 0)], 0, 0, 0)
	for i in range(3):
		load_chunk_sync(i)
	apply_tiles()

func _process(delta: float) -> void:
	var frame_per_chunk = CHUNK_SIZE.y * TILE_SIZE_PIX/ (game_instant.scroll_speed * delta)
	var chunk_size = CHUNK_SIZE.x * CHUNK_SIZE.y
	batch_apply_tiles(ceil(chunk_size /  frame_per_chunk) + 1,3000)
	#batch_apply_tiles(ceil(CHUNK_SIZE.x * game_instant.scroll_speed * delta / TILE_SIZE_PIX),3000)
	var index: int = get_current_chunk_index()
	#print("time: {} delta: {}".format([Time.get_ticks_usec()-t0, delta], "{}"))
	if index != debug_last_index:
		print("当前区块: %d, 卷轴: %.1f" % [index, scroll_progress])
		debug_last_index = index
		
	var current_chunk = get_current_chunk_index()
	terrain_tilemap.position = Vector2(0, scroll_progress)
	if not loaded_chunks.has(current_chunk + 1):
		load_chunk_async(current_chunk + 1)
	if not loaded_chunks.has(current_chunk + 2):
		load_chunk_async(current_chunk + 2)
	if not loaded_chunks.has(current_chunk + 3):
		load_chunk_async(current_chunk + 3)
	if loaded_chunks.has(current_chunk - 5):
		unload_chunk(current_chunk - 5)


func get_visible_chunk_range() -> Vector2i:
	var center_chunk = get_current_chunk_index()
	return Vector2i(center_chunk - 1, center_chunk + 1)

func get_current_chunk_index() -> int:
	var chunk_height = CHUNK_SIZE.y * terrain_tilemap.tile_set.tile_size.y
	var index = int(scroll_progress / chunk_height)
	return clamp(index, 0, 99999) 

func generate_chunk(chunk_ind: int):
	var h_noise = FastNoiseLite.new()
	h_noise.seed = world_seed + 2026
	h_noise.noise_type = FastNoiseLite.TYPE_PERLIN
	h_noise.frequency = 0.1
	
	var t_noise = FastNoiseLite.new()
	t_noise.seed = world_seed - 2026
	t_noise.noise_type = FastNoiseLite.TYPE_PERLIN
	t_noise.frequency = 0.05
	
	var tile_data: Dictionary = {}
	for x in range(CHUNK_SIZE.x):
		for y in range(CHUNK_SIZE.y):
			var world_pos = Vector2i(get_chunk_start_x() + x, get_chunk_start_y(chunk_ind) + y)
			
			var cx = world_pos.x * 0.5
			var cy = world_pos.y * 0.5
			var h = h_noise.get_noise_2d(cx, cy)
			var t = t_noise.get_noise_2d(cx, cy)
			tile_data[world_pos] = calc_terrain_type(h,t)
	var terrain: Dictionary[int, Array] = { 
		0: [],
		1: [],
		2: []
	}
	
	for pos in tile_data:
		if terrain_tilemap.get_cell_source_id(pos) != -1:
			continue
		match tile_data[pos]:
			0: terrain[0].append(pos)  # water
			1, 2: terrain[1].append(pos)  # land
			3, 4: terrain[2].append(pos)  # mud
	if loaded_chunks.has(chunk_ind):
		return
	pending_tiles.append({
		index = chunk_ind,
		terrain = terrain
	})

func load_chunk_sync(chunk_ind: int):
	print("load chunk sync" + str(chunk_ind))
	if loaded_chunks.has(chunk_ind):
		return
	generate_chunk(chunk_ind)
	loaded_chunks[chunk_ind] = true

func load_chunk_async(chunk_index: int) -> void:

	if loaded_chunks.has(chunk_index):
		return
	if pending_chunks.has(chunk_index):
		return
	print("try load chunk async " + str(pending_chunks) + "cur " +str(chunk_index))
	var thread: Thread = Thread.new()
	active_threads.append(thread)
	pending_chunks.append(chunk_index)
	
	thread.start(
		Callable(self, "generate_chunk").bind(chunk_index)
	)
func batch_apply_tiles(batch_size: int, max_time_usec_out: int):
	var t0 = Time.get_ticks_usec()
	var batch_left = batch_size
	while pending_tiles.size():
		var chunk = pending_tiles[0]
		var idx = chunk.index
		var terrain = chunk.terrain as Dictionary[int, Array]
			
		for type_id in terrain.keys():
			var arr: Array = terrain[type_id] as Array[Vector2i]
			if arr.is_empty(): continue
			var t1 = Time.get_ticks_usec()
			var consume = min(batch_left, arr.size())
			var sliced = arr.slice(0, consume)
			terrain_tilemap.set_cells_terrain_connect(sliced, 0, type_id, 0)
			
			var remain = arr.slice(consume, arr.size())	
			arr.clear()
			arr.append_array(remain)
			batch_left -= consume
			var budget_left = max_time_usec_out - (Time.get_ticks_usec() - t0)
			
			print("Set tiles time: " + str(Time.get_ticks_usec() - t1) 
			+ " budget_left: " + str(budget_left)
			+ " batch_size: "+ str(batch_size))
			if batch_left <= 0: continue
			if budget_left <= 0: return
		
		if batch_left >= 0:
			pending_tiles.pop_at(0)
			loaded_chunks[idx] = true
			pending_chunks.erase(idx)
			# clean thread
			for i in range(active_threads.size()):
				var th = active_threads[i]
				if not th.is_started():
					th.wait_to_finish()
					active_threads.remove_at(i)
					break
		t0 = Time.get_ticks_usec()
		
# 预留给同步方法的直接填充tiles的，确保游戏开屏不会出现问题。
# 是完全阻塞主线程的
func apply_tiles(): 
	while pending_tiles.size():
		var chunk = pending_tiles[0]
		var idx = chunk.index
		var terrain = chunk.terrain as Dictionary[int, Array]
		for type_id in terrain.keys():
			var arr: Array = terrain[type_id] as Array[Vector2i]
			if arr.is_empty(): continue
			terrain_tilemap.set_cells_terrain_connect(arr, 0, type_id, 0)
		pending_tiles.pop_at(0)
		loaded_chunks[idx] = true
		pending_chunks.erase(idx)

func apply_chunk(idx: int,terrain: Dictionary[int, Array]):
	#print("apply_chunk ", index, " tiles.size()=", tiles.size())
	if loaded_chunks.has(idx):
		return
	pending_tiles.append({
		index = idx,
		terrain = terrain
	})

func unload_chunk(chunk_index: int) -> void:
	if not loaded_chunks.has(chunk_index):
		return
	print("clear chunk " + str(chunk_index))
	var start_x = get_chunk_start_x()
	var start_y = get_chunk_start_y(chunk_index)
	var end_y = start_y + CHUNK_SIZE.y

	for x in range(CHUNK_SIZE.x):
		for y in range(start_y, end_y):
			terrain_tilemap.set_cell(Vector2i(x + start_x, y), -1) 
	
	loaded_chunks.erase(chunk_index)



static func get_chunk_start_x()-> int: 
	return int(-CHUNK_SIZE.x / 2.0)
static func get_chunk_start_y(ind) -> int:
	return -(ind + 1) * CHUNK_SIZE.y

static func calc_terrain_type(h: float, t: float) -> int:
	var type: int = 0
	if h > -0.1:
		type = 0
	else:
		if t > 0.05:
			type = 1 # grass boundry
		elif t > 0.1:
			type = 2 # grass inland
		else:
			if t > -0.05:
				type = 3 # mud inland
			else:
				type = 4 # mud
	return type

运行效果：


经测试，FPS在区块加载的时候，基本不会有任何的损失或者抖动，相较P1部分对比有很大提升。

---

总结

截至目前。我们把整 Chunk 瞬间写瓦片，优化成了按帧 Streaming 小口喂，我们依然把噪声生成的基本算法用异步去执行，每帧按 delta 算出动态 batch_size，时间预算用完即停，交给下一帧去处理。

流式分片加载这套思路，其实本质就行是把大而整的数据拆成时间-空间双维度的小包裹，按消费能力匀速投递，也就是Godot引擎_process函数当中的delta作为消费能力来对地图生成进行拆分/分片。

其实像开放世界的物件，把宝箱、NPC、植被，当成另一种 Tile，噪声只生成坐标与类型，实际实例化走同一套 pending_tiles 队列。类似的，背包里上万条道具、爆炸后几百个碎片、服务器下推的数兆配置表、像素动画的超大贴图、甚至音视频的解码帧，都可以抽象成“待填充的瓦片队列”，但是要保证生产速度 ≥ 消费速度，需要处理好定预算、再拆切片、队列缓冲这之类的关系。

我们的决策必须围绕，预算第一即消费速度，展开。在 _process 的每一帧里，先用delta 或硬件计时器给本次任务量规定时间或数量，把大数据拆成小片。一旦片段处理完毕，或者所属于其的资源用净，把剩余任务原封不动塞回队列，下一帧举需处理。这样实现可感知的微小延迟换取不可感知的帧率暴跌，同时维持性能损耗。

其实像这类做法也有一定的弊端，那就是批次处理过程中的对于数据的解包拆包、大量的循环，判断逻辑提升了代码的复杂性，同时批量直接一次set_cells_terrain_connect比分多个批次、重复的运行效率要高的多。因为直接处理和传递资源包过程中，不会受到其他任务的干扰或者阻塞，就相当于一个人一心二用或者多用，整体上处理效率是上去了，但是单个任务的效率不会很高，同时如果无法预测/获取消费速度，那么就没办法决定生成的速度（godot刚好提供了delta解决了这个问题）。